Make index names configurable (#467)

Author: cbovis

src/consumers/index-fill-protocol-fee.js

@@ -1,8 +1,11 @@
 const _ = require('lodash');
 const mongoose = require('mongoose');
 
+const { getModel } = require('../model');
 const { JOB, QUEUE } = require('../constants');
+const createDocument = require('../index/fills/create-document');
 const elasticsearch = require('../util/elasticsearch');
+const getIndexName = require('../index/get-index-name');
 
 const indexFillProtocolFee = async (job, { logger }) => {
   const { fillId, protocolFee } = job.data;
@@ -15,23 +18,27 @@ const indexFillProtocolFee = async (job, { logger }) => {
     throw new Error(`Invalid value: ${protocolFee}`);
   }
 
-  const exists = await elasticsearch
-    .getClient()
-    .exists({ id: fillId, index: 'fills', _source: false });
-  const indexed = exists.body;
+  const fill = await getModel('Fill')
+    .findOne({ _id: fillId })
+    .populate([
+      { path: 'takerMetadata', select: 'isContract' },
+      { path: 'transaction', select: 'from' },
+    ])
+    .lean();
 
-  if (!indexed) {
-    throw new Error(`Could not index protocol fee of fill: ${fillId}`);
+  if (fill === null) {
+    throw new Error(`Could not find fill: ${fillId}`);
   }
 
   await elasticsearch.getClient().update({
     id: fillId,
-    index: 'fills',
+    index: getIndexName('fills'),
     body: {
       doc: {
         protocolFeeUSD: protocolFee,
         updatedAt: new Date(Date.now()).toISOString(),
       },
+      upsert: createDocument(fill),
     },
   });
 

src/consumers/index-fill-traders/index.js

@@ -4,6 +4,7 @@ const { JOB, QUEUE } = require('../../constants');
 const { publishJob } = require('../../queues');
 const elasticsearch = require('../../util/elasticsearch');
 const getAddressMetadata = require('../../addresses/get-address-metadata');
+const getIndexName = require('../../index/get-index-name');
 const getTransactionByHash = require('../../transactions/get-transaction-by-hash');
 
 const indexFillTraders = async (job, { logger }) => {
@@ -94,7 +95,7 @@ const indexFillTraders = async (job, { logger }) => {
 
   const result = await elasticsearch
     .getClient()
-    .bulk({ body: `${requestBody}\n`, index: 'trader_fills' });
+    .bulk({ body: `${requestBody}\n`, index: getIndexName('trader_fills') });
 
   if (result.body.errors === true) {
     const errorMessage = _.get(

src/consumers/index-fill-value.js

@@ -5,6 +5,7 @@ const { JOB, QUEUE } = require('../constants');
 const { getModel } = require('../model');
 const createDocument = require('../index/fills/create-document');
 const elasticsearch = require('../util/elasticsearch');
+const getIndexName = require('../index/get-index-name');
 const relayerRegistry = require('../relayers/relayer-registry');
 
 const isOrderMatcher = relayerId => {
@@ -48,7 +49,7 @@ const indexFillValue = async (job, { logger }) => {
 
   await elasticsearch.getClient().update({
     id: fillId,
-    index: 'fills',
+    index: getIndexName('fills'),
     body: {
       doc: {
         tradeVolume: calculateTradeVolume(value, relayerId),

src/consumers/index-fill.js

@@ -2,8 +2,9 @@ const mongoose = require('mongoose');
 
 const { JOB, QUEUE } = require('../constants');
 const { getModel } = require('../model');
-const fillsIndex = require('../index/fills');
 const elasticsearch = require('../util/elasticsearch');
+const fillsIndex = require('../index/fills');
+const getIndexName = require('../index/get-index-name');
 
 const indexFill = async (job, { logger }) => {
   const { fillId } = job.data;
@@ -27,7 +28,7 @@ const indexFill = async (job, { logger }) => {
 
   await elasticsearch.getClient().index({
     id: fill._id,
-    index: 'fills',
+    index: getIndexName('fills'),
     body: fillsIndex.createDocument(fill),
   });
 

src/consumers/index-traded-tokens.js

@@ -2,6 +2,7 @@ const _ = require('lodash');
 
 const { JOB, QUEUE } = require('../constants');
 const elasticsearch = require('../util/elasticsearch');
+const getIndexName = require('../index/get-index-name');
 
 const consumer = async (job, { logger }) => {
   const { attributions, date, fillId, relayerId, tradedTokens } = job.data;
@@ -40,7 +41,7 @@ const consumer = async (job, { logger }) => {
 
   const result = await elasticsearch
     .getClient()
-    .bulk({ body: `${body}\n`, index: 'traded_tokens' });
+    .bulk({ body: `${body}\n`, index: getIndexName('traded_tokens') });
 
   if (result.body.errors === true) {
     const errorMessage = _.get(

src/index/get-index-name.js

@@ -0,0 +1,17 @@
+const getIndexName = index => {
+  if (index === 'fills') {
+    return process.env.INDEX_NAME_FILLS || 'fills';
+  }
+
+  if (index === 'traded_tokens') {
+    return process.env.INDEX_NAME_TRADED_TOKENS || 'traded_tokens';
+  }
+
+  if (index === 'trader_fills') {
+    return process.env.INDEX_NAME_TRADER_FILLS || 'trader_fills';
+  }
+
+  throw new Error(`Unsupported index: ${index}`);
+};
+
+module.exports = getIndexName;