Restructure event extractor for extensibility (#369)

* Restructure event extractor for extensibility

* Tweak a few small things

Author: cbovis

packages/core/src/jobs/extract-events.js

@@ -1,6 +1,6 @@
-const extractorV1 = require('@0x-event-extractor/extractor-v1');
-const extractorV2 = require('@0x-event-extractor/extractor-v2');
-const extractorV3 = require('@0x-event-extractor/extractor-v3');
+const fillExtractorV1 = require('@0x-event-extractor/extractor-v1');
+const fillExtractorV2 = require('@0x-event-extractor/extractor-v2');
+const fillExtractorV3 = require('@0x-event-extractor/extractor-v3');
 
 const { getLogger } = require('../util/logging');
 const BlockRange = require('../model/block-range');
@@ -9,19 +9,19 @@ const getCurrentBlock = require('../ethereum/get-current-block');
 const getNextBlockRange = require('../events/get-next-block-range');
 const withTransaction = require('../util/with-transaction');
 
-const extractEventsForProtocol = async (protocolVersion, extractorConfig) => {
-  // Scope all logging for the job to the specified protocol version
-  const logger = getLogger(`extract events v${protocolVersion}`);
-
-  // Determine which blocks we should fetch log entries from
-  const { currentBlock, fetchLogEntries, getEventData } = extractorConfig;
+const performExtraction = async (currentBlock, extractorConfig) => {
+  const {
+    eventType, // TODO: Query by this once we have data in blockranges collection
+    fetchLogEntries,
+    getEventData,
+    protocolVersion,
+  } = extractorConfig;
 
-  logger.info(`current block is ${currentBlock}`);
+  // Scope all logging for the job to the specified protocol version and event type
+  const logger = getLogger(`extract v${protocolVersion} ${eventType} events`);
 
-  const nextBlockRange = await getNextBlockRange({
-    currentBlock,
-    protocolVersion,
-  });
+  const rangeConfig = { currentBlock, protocolVersion }; // TODO: Also include event type
+  const nextBlockRange = await getNextBlockRange(rangeConfig);
 
   if (nextBlockRange === null) {
     logger.info('no more blocks to process');
@@ -30,53 +30,60 @@ const extractEventsForProtocol = async (protocolVersion, extractorConfig) => {
 
   const { fromBlock, toBlock } = nextBlockRange;
 
-  logger.info(`fetching events from block range ${fromBlock} to ${toBlock}`);
+  logger.info(`fetching events from blocks ${fromBlock}-${toBlock}`);
 
   const logEntries = await fetchLogEntries(fromBlock, toBlock);
+  const entryCount = logEntries.length;
 
-  if (logEntries.length === 0) {
-    logger.info(`no events found in block range ${fromBlock} to ${toBlock}`);
+  if (entryCount === 0) {
+    logger.info(`no events found in blocks ${fromBlock}-${toBlock}`);
   } else {
-    logger.info(
-      `${logEntries.length} events found in block range ${fromBlock} to ${toBlock}`,
-    );
+    logger.info(`${entryCount} events found in blocks ${fromBlock}-${toBlock}`);
   }
 
-  // Persistence operations are wrapped in a transaction to ensure consistency
+  /**
+   * Persistence operations are wrapped in a transaction to ensure consistency
+   * between the BlockRange and Event collections. If a document exists within
+   * the BlockRange collection, then we can assume that all the associated
+   * events will exist in the Event collection.
+   */
   await withTransaction(async session => {
     if (logEntries.length > 0) {
-      // Map the log entries to a common model before persisting to MongoDB
       const events = logEntries.map(logEntry => ({
         blockNumber: logEntry.blockNumber,
         data: getEventData(logEntry),
+        dateIngested: new Date(),
         logIndex: logEntry.logIndex,
         protocolVersion,
         transactionHash: logEntry.transactionHash,
-        type: logEntry.event,
+        type: eventType,
       }));
 
       await Event.insertMany(events, { session });
     }
 
-    // Log details of the queried block range so that we know where
-    // to start from in the next iteration.
-    await BlockRange.findOneAndUpdate(
-      { fromBlock, protocolVersion, toBlock },
-      {
-        $set: {
-          date: new Date(),
+    /**
+     * Log details of the queried block range in MongoDB. This provides
+     * two functions:
+     *
+     * 1. History of scraping activity for debugging
+     * 2. An indicator for where to scrape from on the next iteration
+     *
+     * If collection size became an issue then the BlockRange collection
+     * could be safely capped at say 100,000 documents.
+     */
+    await BlockRange.create(
+      [
+        {
+          dateProcessed: new Date(),
           events: logEntries.length,
+          eventType,
           fromBlock,
           protocolVersion,
           toBlock,
         },
-      },
-      {
-        upsert: true,
-        new: true,
-        runValidators: true,
-        session,
-      },
+      ],
+      { session },
     );
   });
 
@@ -86,13 +93,24 @@ const extractEventsForProtocol = async (protocolVersion, extractorConfig) => {
 };
 
 const extractEvents = async () => {
+  const logger = getLogger('event extractor');
+
+  logger.info('beginning event extraction');
+  logger.info('fetching current block');
+
   const currentBlock = await getCurrentBlock();
 
-  // Fetching of events is delegated to version specific extractors which
-  // interact with the correct 0x SDK version and contract.
-  await extractEventsForProtocol(1, { currentBlock, ...extractorV1 });
-  await extractEventsForProtocol(2, { currentBlock, ...extractorV2 });
-  await extractEventsForProtocol(3, { currentBlock, ...extractorV3 });
+  logger.info(`current block is ${currentBlock}`);
+
+  /**
+   * Extractors are run sequentially to help avoid issues with rate
+   * limiting in the Ethereum RPC provider.
+   */
+  await performExtraction(currentBlock, fillExtractorV1);
+  await performExtraction(currentBlock, fillExtractorV2);
+  await performExtraction(currentBlock, fillExtractorV3);
+
+  logger.info('finished event extraction');
 };
 
 module.exports = extractEvents;

packages/core/src/model/block-range.js

@@ -3,21 +3,23 @@ const mongoose = require('mongoose');
 const { Schema } = mongoose;
 
 const schema = Schema({
-  dateProcessed: Date,
-  events: Number,
-  fromBlock: Number,
-  protocolVersion: { default: 1, type: Number },
-  toBlock: Number,
+  dateProcessed: { required: true, type: Date },
+  events: { required: true, type: Number },
+  eventType: { required: true, type: String },
+  fromBlock: { required: true, type: Number },
+  protocolVersion: { default: 1, required: true, type: Number },
+  toBlock: { required: true, type: Number },
 });
 
 // Used for determining last processed block
 schema.index({ protocolVersion: 1, toBlock: -1 });
 
+// Used to enforce consistency in the data
 schema.index(
   { fromBlock: 1, toBlock: 1, protocolVersion: 1 },
   { unique: true },
 );
 
-const Model = mongoose.model('BlockRange', schema);
+const BlockRange = mongoose.model('BlockRange', schema);
 
-module.exports = Model;
+module.exports = BlockRange;

packages/core/src/model/event.js

@@ -3,18 +3,18 @@ const mongoose = require('mongoose');
 const { Schema } = mongoose;
 
 const schema = Schema({
-  blockNumber: Number,
-  data: { type: Schema.Types.Mixed },
-  fillCreated: { type: Boolean, default: false },
-  logIndex: Number,
-  protocolVersion: { default: 1, type: Number },
-  transactionHash: String,
-  type: String,
+  blockNumber: { required: true, type: Number },
+  data: { required: true, type: Schema.Types.Mixed },
+  dateIngested: { required: true, type: Date },
+  logIndex: { required: true, type: Number },
+  protocolVersion: { default: 1, required: true, type: Number },
+  transactionHash: { required: true, type: String },
+  type: { required: true, type: String },
 });
 
+// Used to enforce consistency in the data
 schema.index({ logIndex: 1, transactionHash: 1 }, { unique: true });
-schema.index({ fillCreated: 1 });
 
-const Model = mongoose.model('Event', schema);
+const Event = mongoose.model('Event', schema);
 
-module.exports = Model;
+module.exports = Event;

packages/core/src/model/index.js

@@ -2,8 +2,8 @@ const BlockRange = require('./block-range');
 const Event = require('./event');
 
 const init = async () => {
-  BlockRange.createCollection();
-  Event.createCollection();
+  await BlockRange.createCollection();
+  await Event.createCollection();
 };
 
 module.exports = { BlockRange, Event, init };

packages/extractor-v1/package.json

@@ -1,5 +1,6 @@
 {
   "name": "@0x-event-extractor/extractor-v1",
+  "description": "Extractor for v1 LogFill events",
   "version": "1.0.0",
   "main": "src/index.js",
   "dependencies": {

packages/extractor-v1/src/index.js

@@ -6,4 +6,10 @@ const configure = options => {
   zeroEx.configure(options);
 };
 
-module.exports = { configure, fetchLogEntries, getEventData };
+module.exports = {
+  configure,
+  eventType: 'LogFill',
+  fetchLogEntries,
+  getEventData,
+  protocolVersion: 1,
+};

packages/extractor-v2/package.json

@@ -1,5 +1,6 @@
 {
   "name": "@0x-event-extractor/extractor-v2",
+  "description": "Extractor for v2 Fill events",
   "version": "1.0.0",
   "main": "src/index.js",
   "dependencies": {

packages/extractor-v2/src/index.js

@@ -6,4 +6,10 @@ const configure = options => {
   zeroEx.configure(options);
 };
 
-module.exports = { configure, fetchLogEntries, getEventData };
+module.exports = {
+  configure,
+  eventType: 'Fill',
+  fetchLogEntries,
+  getEventData,
+  protocolVersion: 2,
+};

packages/extractor-v3/package.json

@@ -1,5 +1,6 @@
 {
   "name": "@0x-event-extractor/extractor-v3",
+  "description": "Extractor for v3 Fill events",
   "version": "1.0.0",
   "main": "src/index.js",
   "dependencies": {

packages/extractor-v3/src/index.js

@@ -6,4 +6,10 @@ const configure = options => {
   zeroEx.configure(options);
 };
 
-module.exports = { configure, fetchLogEntries, getEventData };
+module.exports = {
+  configure,
+  eventType: 'Fill',
+  fetchLogEntries,
+  getEventData,
+  protocolVersion: 3,
+};