<commit_before>'use strict';

const util = require('util');
const EventEmitter = require('events').EventEmitter;
const Bluebird = require('bluebird');
const elasticsearch = require('elasticsearch');
const retry = require('bluebird-retry');
const _ = require('lodash');
const errors = require('./errors');

util.inherits(ElasticsearchProvider, EventEmitter);

/**
 * Elasticsearch provider
 * @param {object} [options]
 * @constructor
 */

function ElasticsearchProvider(options) {

  if (options && !_.isPlainObject(options)) {
    throw new errors.InvalidArgumentError('invalid-options');
  }

  if (options && options.bulkSize && !_.isFinite(options.bulkSize)) {
    throw new errors.InvalidArgumentError('invalid-bulk-size');
  }

  if (options && options.bulkTimeout && !_.isFinite(options.bulkTimeout)) {
    throw new errors.InvalidArgumentError('invalid-bulk-timeout');
  }

  if (options && options.bulkBufferSize && !_.isFinite(options.bulkBufferSize)) {
    throw new errors.InvalidArgumentError('invalid-bulk-buffer-size');
  }

  this.bulkBuffer = [];
  this.bulkTimer = null;
  this.bulkSize = options && options.bulkSize ? options.bulkSize : 2000;
  this.bulkBufferSize = options && options.bulkBufferSize ? options.bulkBufferSize : 4000;
  this.bulkTimeout = options && options.bulkTimeout ? options.bulkTimeout : 10000;
  this.isFlushingBulkBuffer = false;
  this.requestTimeout = 1000;
  this.client = null;
}

/**
 * Connects to Elasticsearch and tests the connection with a ping
 * Returns promise or calls callback (if provided)
 * @param {string|array} hosts
 * @param {function} [callback]
 * @returns {Promise}
 */

ElasticsearchProvider.prototype.connect = function(hosts, callback) {

  return Bluebird.resolve()
    .then(() => {

      if (!this.client) {
        this.client = new elasticsearch.Client({hosts});
      }

      return this.client.ping({
        requestTimeout: this.requestTimeout,
        hello: 'mongoolastic'
      });
    })
    .nodeify(callback);
};

/**
 * Indexes a mongoose document in Elasticsearch
 * Returns a promise or calls callback (if provided)
 * @param {string} id
 * @param {object} doc
 * @param {string} type
 * @param {string} index
 * @param {boolean} useBulk
 * @param {function} [callback]
 * @throws {InvalidArgumentError}
 * @returns {Promise}
 */

ElasticsearchProvider.prototype.indexDoc = function(id, doc, type, index, useBulk, callback) {

  return Bluebird.resolve([id, doc, type, index])
    .spread((id, doc, type, index) => {

      if (!this.isValidId(id)) {
        throw new errors.InvalidArgumentError('invalid-id');
      }

      if (!this.isValidType(type)) {
        throw new errors.InvalidArgumentError('invalid-type');
      }

      if (!this.isValidIndex(index)) {
        throw new errors.InvalidArgumentError('invalid-index-name');
      }

      if (useBulk) {

        const action = {index: {_index: index, _type: type, _id: id}};
        return this.addToBulkBuffer(action, doc);

      } else {

        return this.client.index({
          index,
          type,
          id,
          body: doc,
          refresh: true
        });
      }
    })
    .nodeify(callback);
};

/**
 * Flushes the buffer and performs bulk insert/delete
 * Because of it's decoupled nature, errors are emitted as events
 */

ElasticsearchProvider.prototype.flushBulkBuffer = function() {

  if (this.bulkTimer) {
    clearTimeout(this.bulkTimer);
    this.bulkTimer = null;
  }

  if (this.isFlushingBulkBuffer) {
    return;
  }

  this.isFlushingBulkBuffer = true;

  const body = [];<commit_msg>Bulk indexing OK<commit_after>'use strict';

const util = require('util');
const EventEmitter = require('events').EventEmitter;
const Bluebird = require('bluebird');
const elasticsearch = require('elasticsearch');
const retry = require('bluebird-retry');
const _ = require('lodash');
const errors = require('./errors');

util.inherits(ElasticsearchProvider, EventEmitter);

/**
 * Elasticsearch provider
 * @param {object} [options]
 * @constructor
 */

function ElasticsearchProvider(options) {

  if (options && !_.isPlainObject(options)) {
    throw new errors.InvalidArgumentError('invalid-options');
  }

  if (options && options.bulkSize && !_.isFinite(options.bulkSize)) {
    throw new errors.InvalidArgumentError('invalid-bulk-size');
  }

  if (options && options.bulkTimeout && !_.isFinite(options.bulkTimeout)) {
    throw new errors.InvalidArgumentError('invalid-bulk-timeout');
  }

  if (options && options.bulkBufferSize && !_.isFinite(options.bulkBufferSize)) {
    throw new errors.InvalidArgumentError('invalid-bulk-buffer-size');
  }

  this.bulkBuffer = [];
  this.bulkTimer = null;
  this.bulkSize = options && options.bulkSize ? options.bulkSize : 2000;
  this.bulkBufferSize = options && options.bulkBufferSize ? options.bulkBufferSize : 4000;
  this.bulkTimeout = options && options.bulkTimeout ? options.bulkTimeout : 10000;
  this.isFlushingBulkBuffer = false;
  this.requestTimeout = 1000;
  this.client = null;
}

/**
 * Connects to Elasticsearch and tests the connection with a ping
 * Returns promise or calls callback (if provided)
 * @param {string|array} hosts
 * @param {function} [callback]
 * @returns {Promise}
 */

ElasticsearchProvider.prototype.connect = function(hosts, callback) {

  return Bluebird.resolve()
    .then(() => {

      if (!this.client) {
        this.client = new elasticsearch.Client({hosts});
      }

      return this.client.ping({
        requestTimeout: this.requestTimeout,
        hello: 'mongoolastic'
      });
    })
    .nodeify(callback);
};

/**
 * Indexes a mongoose document in Elasticsearch
 * Returns a promise or calls callback (if provided)
 * @param {string} id
 * @param {object} doc
 * @param {string} type
 * @param {string} index
 * @param {boolean} useBulk
 * @param {function} [callback]
 * @throws {InvalidArgumentError}
 * @returns {Promise}
 */

ElasticsearchProvider.prototype.indexDoc = function(id, doc, type, index, useBulk, callback) {

  return Bluebird.resolve([id, doc, type, index])
    .spread((id, doc, type, index) => {

      if (!this.isValidId(id)) {
        throw new errors.InvalidArgumentError('invalid-id');
      }

      if (!this.isValidType(type)) {
        throw new errors.InvalidArgumentError('invalid-type');
      }

      if (!this.isValidIndex(index)) {
        throw new errors.InvalidArgumentError('invalid-index-name');
      }

      if (useBulk) {

        const action = {index: {_index: index, _type: type, _id: id}};
        return this.addToBulkBuffer(action, doc);

      } else {

        return this.client.index({
          index,
          type,
          id,
          body: doc,
          refresh: true
        });
      }
    })
    .nodeify(callback);
};

/**
 * Flushes the buffer and performs bulk insert/delete
 * Because of it's decoupled nature, errors are emitted as events
 * @emits {Error}
 */

ElasticsearchProvider.prototype.flushBulkBuffer = function() {

  if (this.bulkTimer) {
    clearTimeout(this.bulkTimer);
    this.bulkTimer = null;
  }

  if (this.isFlushingBulkBuffer) {
    return;
  }

  this.isFlushingBulkBuffer = true;

  const body = [];
{'fraction_non_alphanumeric': 0.13476263399693722, 'fraction_numerical': 0.004733398301545315, 'mean_word_length': 3.4427952999381572, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '2444432', 'n_tokens_mistral': 2217, 'n_tokens_neox': 2157, 'n_words': 502}