weseek__growi/lib/util/search.js

/**
 * Search
 */

var elasticsearch = require('elasticsearch'),
  debug = require('debug')('crowi:lib:search');

function SearchClient(crowi, esUri) {
  this.DEFAULT_OFFSET = 0;
  this.DEFAULT_LIMIT = 50;

  this.esUri = esUri;
  this.crowi = crowi;

  var uri = this.parseUri(this.esUri);
  this.host = uri.host;
  this.index_name = uri.index_name;

  this.client = new elasticsearch.Client({
    host: this.host,
    requestTimeout: 5000,
  });

  this.registerUpdateEvent();

  this.mappingFile = crowi.resourceDir + 'search/mappings.json';
}

SearchClient.prototype.registerUpdateEvent = function() {
  var pageEvent = this.crowi.event('page');
  pageEvent.on('create', this.syncPageCreated.bind(this))
  pageEvent.on('update', this.syncPageUpdated.bind(this))
};

SearchClient.prototype.parseUri = function(uri) {
  if (!(m = uri.match(/^elasticsearch:\/\/([^:]+):([^\/]+)\/(.+)$/))) {
    throw new Error('Invalid ELASTICSEARCH_URI format. Should be elasticsearch://host:port/index_name');
  }

  return {
    host: m[1] + ':' + m[2],
    index_name: m[3],
  };
};

SearchClient.prototype.buildIndex = function(uri) {
  return this.client.indices.create({
    index: this.index_name,
    body: require(this.mappingFile)
  });
};

SearchClient.prototype.deleteIndex = function(uri) {
  return this.client.indices.delete({
    index: this.index_name,
  });
};

SearchClient.prototype.prepareBodyForUpdate = function(body, page) {
  if (!Array.isArray(body)) {
    throw new Error('Body must be an array.');
  }

  var command = {
    update: {
      _index: this.index_name,
      _type: 'pages',
      _id: page._id.toString(),
    }
  };

  var document = {
    doc: {
      path: page.path,
      body: page.revision.body,
      comment_count: page.commentCount,
      bookmark_count: 0, // todo
      like_count: page.liker.length || 0,
      updated_at: page.updatedAt,
    }
  };

  body.push(command);
  body.push(document);
};

SearchClient.prototype.prepareBodyForCreate = function(body, page) {
  if (!Array.isArray(body)) {
    throw new Error('Body must be an array.');
  }

  var command = {
    index: {
      _index: this.index_name,
      _type: 'pages',
      _id: page._id.toString(),
    }
  };

  var document = {
    path: page.path,
    body: page.revision.body,
    username: page.creator.username,
    comment_count: page.commentCount,
    bookmark_count: 0, // todo
    like_count: page.liker.length || 0,
    created_at: page.createdAt,
    updated_at: page.updatedAt,
  };

  body.push(command);
  body.push(document);
};

SearchClient.prototype.addPages = function(pages)
{
  var self = this;
  var body = [];

  pages.map(function(page) {
    self.prepareBodyForCreate(body, page);
  });

  debug('addPages(): Sending Request to ES', body);
  return this.client.bulk({
    body: body,
  });
};

SearchClient.prototype.updatePages = function(pages)
{
  var self = this;
  var body = [];

  pages.map(function(page) {
    self.prepareBodyForUpdate(body, page);
  });

  debug('updatePages(): Sending Request to ES', body);
  return this.client.bulk({
    body: body,
  });
};

SearchClient.prototype.addAllPages = function()
{
  var self = this;
  var offset = 0;
  var Page = this.crowi.model('Page');
  var stream = Page.getStreamOfFindAll();
  var body = [];

  return new Promise(function(resolve, reject) {
    stream.on('data', function (doc) {
      if (!doc.creator || !doc.revision) {
        debug('Skipped', doc.path);
        return ;
      }

      self.prepareBodyForCreate(body, doc);
    }).on('error', function (err) {
      // TODO: handle err
      debug('Error stream:', err);
    }).on('close', function () {
      // all done

      // 最後に送信
      self.client.bulk({ body: body, })
      .then(function(res) {
        debug('Reponse from es:', res);
        return resolve(res);
      }).catch(function(err) {
        debug('Err from es:', err);
        return reject(err);
      });
    });
  });
};

/**
 * search returning type:
 * {
 *   meta: { total: Integer, results: Integer},
 *   data: [ pages ...],
 * }
 */
SearchClient.prototype.search = function(query)
{
  var self = this;

  return new Promise(function(resolve, reject) {
    self.client.search(query)
    .then(function(data) {
      var result = {
        meta: {
          took: data.took,
          total: data.hits.total,
          results: data.hits.hits.length,
        },
        data: data.hits.hits.map(function(elm) {
          return {_id: elm._id, _score: elm._score};
        })
      };

      resolve(result);
    }).catch(function(err) {
      reject(err);
    });
  });
};

SearchClient.prototype.createSearchQuerySortedByUpdatedAt = function(option)
{
  // getting path by default is almost for debug
  var fields = ['path', '_id'];
  if (option) {
    fields = option.fields || fields;
  }

  // default is only id field, sorted by updated_at
  var query = {
    index: this.index_name,
    type: 'pages',
    body: {
      fields: fields,
      sort: [{ updated_at: { order: 'desc'}}],
      query: {}, // query
    }
  };
  this.appendResultSize(query);

  return query;
};

SearchClient.prototype.createSearchQuerySortedByScore = function(option)
{
  var fields = ['path', '_id'];
  if (option) {
    fields = option.fields || fields;
  }

  // sort by score
  var query = {
    index: this.index_name,
    type: 'pages',
    body: {
      fields: fields,
      sort: [ {_score: { order: 'desc'} }],
      query: {}, // query
    }
  };
  this.appendResultSize(query);

  return query;
};

SearchClient.prototype.appendResultSize = function(query, from, size)
{
  query.from = from || this.DEFAULT_OFFSET;
  query.size = size || this.DEFAULT_LIMIT;
};

SearchClient.prototype.appendCriteriaForKeywordContains = function(query, keyword)
{
  // query is created by createSearchQuerySortedByScore() or createSearchQuerySortedByUpdatedAt()
  if (!query.body.query.bool) {
    query.body.query.bool = {};
  }

  if (!query.body.query.bool.must || !Array.isArray(query.body.query.must)) {
    query.body.query.bool.must = [];
  }

  query.body.query.bool.must.push({
    multi_match: {
      query: keyword,
      fields: [
        "path.ja^2", // ためしに。
        "body.ja"
      ],
      operator: "and"
    }
  });
};

SearchClient.prototype.appendCriteriaForPathFilter = function(query, path)
{
  // query is created by createSearchQuerySortedByScore() or createSearchQuerySortedByUpdatedAt()
  if (!query.body.query.bool) {
    query.body.query.bool = {};
  }

  if (!query.body.query.bool.filter || !Array.isArray(query.body.query.bool.filter)) {
    query.body.query.bool.filter = [];
  }

  if (path.match(/\/$/)) {
    path = path.substr(0, path.length - 1);
  }
  query.body.query.bool.filter.push({
    wildcard: {
      "path.raw": path + "/*"
    }
  });
};

SearchClient.prototype.searchKeyword = function(keyword, option)
{
  var from = option.offset || null;
  var query = this.createSearchQuerySortedByScore();
  this.appendCriteriaForKeywordContains(query, keyword);

  return this.search(query);
};

SearchClient.prototype.searchByPath = function(keyword, prefix)
{
  // TODO path 名だけから検索
};

SearchClient.prototype.searchKeywordUnderPath = function(keyword, path, option)
{
  var from = option.offset || null;
  var query = this.createSearchQuerySortedByScore();
  this.appendCriteriaForKeywordContains(query, keyword);
  this.appendCriteriaForPathFilter(query, path);

  if (from) {
    this.appendResultSize(query, from);
  }

  return this.search(query);
};

SearchClient.prototype.syncPageCreated = function(page, user)
{
  this.addPages([page])
  .then(function(res) {
    debug('ES Response', res);
  })
  .catch(function(err){
    debug('ES Error', err);
  });
};

SearchClient.prototype.syncPageUpdated = function(page, user)
{
  this.updatePages([page])
  .then(function(res) {
    debug('ES Response', res);
  })
  .catch(function(err){
    debug('ES Error', err);
  });
};


module.exports = SearchClient;