Start indexing fill value in Elasticsearch (#373)

* Start indexing fill value

* Ensure value is indexed when reindexing fills

Author: cbovis

src/constants.js

@@ -34,6 +34,7 @@ module.exports = {
     FETCH_FILL_STATUS: 'fetch-fill-status',
     INDEX_FILL: 'index-fill',
     INDEX_FILL_STATUS: 'index-fill-status',
+    INDEX_FILL_VALUE: 'index-fill-value',
   },
   QUEUE: {
     BULK_INDEXING: 'bulk-indexing',

src/consumers/index-fill-value.js

@@ -0,0 +1,43 @@
+const _ = require('lodash');
+const mongoose = require('mongoose');
+const signale = require('signale');
+
+const { JOB, QUEUE } = require('../constants');
+const elasticsearch = require('../util/elasticsearch');
+
+const logger = signale.scope('index fill value');
+
+const indexFillValue = async job => {
+  const { fillId, value } = job.data;
+
+  if (!mongoose.Types.ObjectId.isValid(fillId)) {
+    throw new Error(`Invalid fillId: ${fillId}`);
+  }
+
+  if (!_.isFinite(value)) {
+    throw new Error(`Invalid value: ${value}`);
+  }
+
+  const exists = await elasticsearch
+    .getClient()
+    .exists({ id: fillId, index: 'fills', _source: false });
+  const indexed = exists.body;
+
+  if (!indexed) {
+    throw new Error(`Could not update value of fill: ${fillId}`);
+  }
+
+  await elasticsearch.getClient().update({
+    id: fillId,
+    index: 'fills',
+    body: { doc: { value } },
+  });
+
+  logger.success(`indexed fill value: ${fillId}`);
+};
+
+module.exports = {
+  fn: indexFillValue,
+  jobName: JOB.INDEX_FILL_VALUE,
+  queueName: QUEUE.FILL_INDEXING,
+};

src/consumers/index.js

@@ -4,8 +4,15 @@ const bulkIndexFills = require('./bulk-index-fills');
 const fetchFillStatus = require('./fetch-fill-status');
 const indexFill = require('./index-fill');
 const indexFillStatus = require('./index-fill-status');
+const indexFillValue = require('./index-fill-value');
 
-const consumers = [bulkIndexFills, fetchFillStatus, indexFill, indexFillStatus];
+const consumers = [
+  bulkIndexFills,
+  fetchFillStatus,
+  indexFill,
+  indexFillStatus,
+  indexFillValue,
+];
 
 const initQueueConsumers = config => {
   const queues = getQueues();

src/fixtures/fills/v2.js

@@ -0,0 +1,49 @@
+const V2_FILL = {
+  _id: '5e01056923573c61d846f51d',
+  conversions: { USD: { makerFee: 0, takerFee: 0, amount: 658.8957691929245 } },
+  hasValue: true,
+  immeasurable: false,
+  status: 1,
+  assets: [
+    {
+      tokenResolved: true,
+      _id: '5e01056923573c61d846f51f',
+      amount: 660879036587289100000,
+      tokenAddress: '0x6b175474e89094c44da98b954eedeac495271d0f',
+      actor: 0,
+      price: { USD: 0.9969990462935456 },
+      value: { USD: 658.8957691929245 },
+    },
+    {
+      tokenResolved: true,
+      _id: '5e01056923573c61d846f51e',
+      amount: 5063754758629915000,
+      tokenAddress: '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2',
+      actor: 1,
+      price: { USD: 130.12 },
+      value: { USD: 658.8957691929245 },
+    },
+  ],
+  blockHash:
+    '0x7c19236809f502da3d481761009377877171ccd8757df293f7537c974674c81b',
+  blockNumber: 9151911,
+  date: '2019-12-23T18:15:16.000Z',
+  eventId: '5e01053a3e349627a0408d51',
+  fees: [],
+  feeRecipient: '0xa258b39954cef5cb142fd567a46cddb31a670124',
+  logIndex: 38,
+  maker: '0x6924a03bb710eaf199ab6ac9f2bb148215ae9b5d',
+  makerFee: 0,
+  orderHash:
+    '0xf186241e774ed0ecb6b6c101482f9a52d77caaf2db3165c9052f814e4c844f5e',
+  protocolVersion: 2,
+  relayerId: 7,
+  senderAddress: '0x8018280076d7fa2caa1147e441352e8a89e1ddbe',
+  taker: '0x8018280076d7fa2caa1147e441352e8a89e1ddbe',
+  takerFee: 0,
+  transactionHash:
+    '0x36a35ae64def8beab677435be4762bf63977000172d47096230a94922383c856',
+  pricingStatus: 0,
+};
+
+module.exports = V2_FILL;

src/index/fills/__snapshots__/create-document.test.js.snap

@@ -0,0 +1,26 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`should create Elasticsearch document for fill 1`] = `
+Object {
+  "assets": Array [
+    Object {
+      "tokenAddress": "0x6b175474e89094c44da98b954eedeac495271d0f",
+    },
+    Object {
+      "tokenAddress": "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
+    },
+  ],
+  "date": "2019-12-23T18:15:16.000Z",
+  "feeRecipient": "0xa258b39954cef5cb142fd567a46cddb31a670124",
+  "fees": Array [],
+  "maker": "0x6924a03bb710eaf199ab6ac9f2bb148215ae9b5d",
+  "orderHash": "0xf186241e774ed0ecb6b6c101482f9a52d77caaf2db3165c9052f814e4c844f5e",
+  "protocolVersion": 2,
+  "relayerId": 7,
+  "senderAddress": "0x8018280076d7fa2caa1147e441352e8a89e1ddbe",
+  "status": 1,
+  "taker": "0x8018280076d7fa2caa1147e441352e8a89e1ddbe",
+  "transactionHash": "0x36a35ae64def8beab677435be4762bf63977000172d47096230a94922383c856",
+  "value": 658.8957691929245,
+}
+`;

src/index/fills/create-document.js

@@ -1,16 +1,23 @@
-const createDocument = fill => ({
-  assets: fill.assets.map(asset => ({ tokenAddress: asset.tokenAddress })),
-  date: fill.date,
-  fees: fill.fees.map(fee => ({ tokenAddress: fee.tokenAddress })),
-  feeRecipient: fill.feeRecipient,
-  maker: fill.maker,
-  orderHash: fill.orderHash,
-  protocolVersion: fill.protocolVersion,
-  relayerId: fill.relayerId,
-  senderAddress: fill.senderAddress,
-  status: fill.status,
-  taker: fill.taker,
-  transactionHash: fill.transactionHash,
-});
+const _ = require('lodash');
+
+const createDocument = fill => {
+  const value = _.get(fill, 'conversions.USD.amount');
+
+  return {
+    assets: fill.assets.map(asset => ({ tokenAddress: asset.tokenAddress })),
+    date: fill.date,
+    fees: fill.fees.map(fee => ({ tokenAddress: fee.tokenAddress })),
+    feeRecipient: fill.feeRecipient,
+    maker: fill.maker,
+    orderHash: fill.orderHash,
+    protocolVersion: fill.protocolVersion,
+    relayerId: fill.relayerId,
+    senderAddress: fill.senderAddress,
+    status: fill.status,
+    taker: fill.taker,
+    transactionHash: fill.transactionHash,
+    value: value === null ? undefined : value,
+  };
+};
 
 module.exports = createDocument;

src/index/fills/create-document.test.js

@@ -0,0 +1,20 @@
+const createDocument = require('./create-document');
+const V2_FILL = require('../../fixtures/fills/v2');
+
+it('should create Elasticsearch document for fill', () => {
+  const doc = createDocument(V2_FILL);
+  expect(doc).toMatchSnapshot();
+});
+
+it('should exclude value property when fill is unmeasured', () => {
+  const fill = {
+    ...V2_FILL,
+    conversions: {
+      ...V2_FILL.conversions,
+      USD: { ...V2_FILL.conversions.USD, amount: null },
+    },
+  };
+  const doc = createDocument(fill);
+
+  expect(doc.value).toBeUndefined();
+});

src/jobs/measure-fills/measure-fill.js

@@ -1,7 +1,9 @@
 const bluebird = require('bluebird');
 const signale = require('signale');
 
+const { JOB, QUEUE } = require('../../constants');
 const { getToken } = require('../../tokens/token-cache');
+const { publishJob } = require('../../queues');
 const formatTokenAmount = require('../../tokens/format-token-amount');
 const getConversionRate = require('../../rates/get-conversion-rate');
 const getMeasurableActor = require('./get-measurable-actor');
@@ -62,6 +64,10 @@ const measureFill = async fill => {
   await withTransaction(async session => {
     await fill.save({ session });
     await persistTokenPrices(tokenPrices, fill, session);
+    await publishJob(QUEUE.FILL_INDEXING, JOB.INDEX_FILL_VALUE, {
+      fillId: fill._id,
+      value: totalValue,
+    });
   });
 
   logger.debug(`set value of fill ${fill._id} to ${totalValue}`);