Re-organize event parsing and decoding

6 files changed, 120 insertions, 89 deletions

diff --git a/packages/pipeline/src/data-sources/etherscan/index.ts b/packages/pipeline/src/data_sources/etherscan/index.ts
index c5eb2b9c6..044fff02e 100644
--- a/packages/pipeline/src/data-sources/etherscan/index.ts
+++ b/packages/pipeline/src/data_sources/etherscan/index.ts
@@ -1,7 +1,5 @@
 import { default as axios } from 'axios';
-import { AbiDefinition, BlockParam, BlockParamLiteral } from 'ethereum-types';
-
-import { EventsResponse, ExchangeEventEntity, parseRawEventsResponse } from './events';
+import { BlockParam, BlockParamLiteral } from 'ethereum-types';
 
 const ETHERSCAN_URL = 'https://api.etherscan.io/api';
 
@@ -12,25 +10,43 @@ export class Etherscan {
     }
 
     /**
-     * Gets the decoded events for a specific contract and block range.
+     * Gets the raw events for a specific contract and block range.
      * @param contractAddress  The address of the contract to get the events for.
-     * @param constractAbi The ABI of the contract.
      * @param fromBlock The start of the block range to get events for (inclusive).
      * @param toBlock The end of the block range to get events for (inclusive).
      * @returns A list of decoded events.
      */
     public async getContractEventsAsync(
         contractAddress: string,
-        contractAbi: AbiDefinition[],
         fromBlock: BlockParam = BlockParamLiteral.Earliest,
         toBlock: BlockParam = BlockParamLiteral.Latest,
-    ): Promise<ExchangeEventEntity[]> {
+    ): Promise<EventsResponse> {
         const fullURL = `${ETHERSCAN_URL}?module=logs&action=getLogs&address=${contractAddress}&fromBlock=${fromBlock}&toBlock=${toBlock}&apikey=${
             this._apiKey
         }`;
         const resp = await axios.get<EventsResponse>(fullURL);
         // TODO(albrow): Check response code.
-        const decodedEvents = parseRawEventsResponse(contractAbi, resp.data);
-        return decodedEvents;
+        return resp.data;
     }
 }
+
+// Raw events response from etherescan.io
+export interface EventsResponse {
+    status: string;
+    message: string;
+    result: EventsResponseResult[];
+}
+
+// Events as represented in the response from etherscan.io
+export interface EventsResponseResult {
+    address: string;
+    topics: string[];
+    data: string;
+    blockNumber: string;
+    timeStamp: string;
+    gasPrice: string;
+    gasUsed: string;
+    logIndex: string;
+    transactionHash: string;
+    transactionIndex: string;
+}
diff --git a/packages/pipeline/src/data_types/events/event_handlers/base_event_handler.ts b/packages/pipeline/src/data_types/events/event_handlers/base_event_handler.ts
new file mode 100644
index 000000000..59331fc4f
--- /dev/null
+++ b/packages/pipeline/src/data_types/events/event_handlers/base_event_handler.ts
@@ -0,0 +1,34 @@
+import { AbiDefinition, BlockParam, BlockParamLiteral, LogEntry } from 'ethereum-types';
+import * as R from 'ramda';
+import { BaseEntity } from 'typeorm';
+
+import { Etherscan } from '../../../data_sources/etherscan';
+import { convertResponseToLogEntry } from '../event_utils';
+
+export abstract class BaseEventHandler<EntityType extends BaseEntity> {
+    protected _abi: AbiDefinition[];
+    protected _address: string;
+    protected _etherscan: Etherscan;
+    constructor(abi: AbiDefinition[], address: string, etherscan: Etherscan) {
+        this._abi = abi;
+        this._address = address;
+        this._etherscan = etherscan;
+    }
+    public abstract convertLogEntryToEventEntity(logEntry: LogEntry): EntityType;
+
+    public async getEventsAsync(
+        fromBlock: BlockParam = BlockParamLiteral.Earliest,
+        toBlock: BlockParam = BlockParamLiteral.Latest,
+    ): Promise<EntityType[]> {
+        const rawEventsResponse = await this._etherscan.getContractEventsAsync(this._address, fromBlock, toBlock);
+        const logEntries = R.map(convertResponseToLogEntry, rawEventsResponse.result);
+        // Note(albrow): Imperative for loop is required here because we can't
+        // bind convertLogEntryToEventEntity without having a specific instance
+        // of a sub-class.
+        const result = [];
+        for (const logEntry of logEntries) {
+            result.push(this.convertLogEntryToEventEntity(logEntry));
+        }
+        return result;
+    }
+}
diff --git a/packages/pipeline/src/data-sources/etherscan/events.ts b/packages/pipeline/src/data_types/events/event_handlers/exchange_event_handler.ts
index a828af527..38ff20595 100644
--- a/packages/pipeline/src/data-sources/etherscan/events.ts
+++ b/packages/pipeline/src/data_types/events/event_handlers/exchange_event_handler.ts
@@ -5,70 +5,28 @@ import { AbiDecoder, BigNumber } from '@0xproject/utils';
 import { AbiDefinition, LogEntry, LogWithDecodedArgs } from 'ethereum-types';
 import * as R from 'ramda';
 
-import { ExchangeFillEvent } from '../../entities/ExchangeFillEvent';
+import { ExchangeFillEvent } from '../../../entities/ExchangeFillEvent';
+import { decodeLogEntry } from '../event_utils';
+
+import { BaseEventHandler } from './base_event_handler';
 
 // TODO(albrow): Union with other exchange event entity types
 export type ExchangeEventEntity = ExchangeFillEvent;
 
-// Raw events response from etherescan.io
-export interface EventsResponse {
-    status: string;
-    message: string;
-    result: EventsResponseResult[];
-}
-
-// Events as represented in the response from etherscan.io
-export interface EventsResponseResult {
-    address: string;
-    topics: string[];
-    data: string;
-    blockNumber: string;
-    timeStamp: string;
-    gasPrice: string;
-    gasUsed: string;
-    logIndex: string;
-    transactionHash: string;
-    transactionIndex: string;
-}
-
-const hexRadix = 16;
-
-function hexToInt(hex: string): number {
-    return parseInt(hex.replace('0x', ''), hexRadix);
-}
-
-// Converts a raw event response to a LogEntry
-// tslint:disable-next-line:completed-docs
-export function _convertResponseToLogEntry(result: EventsResponseResult): LogEntry {
-    return {
-        logIndex: hexToInt(result.logIndex),
-        transactionIndex: hexToInt(result.transactionIndex),
-        transactionHash: result.transactionHash,
-        blockHash: '',
-        blockNumber: hexToInt(result.blockNumber),
-        address: result.address,
-        data: result.data,
-        topics: result.topics,
-    };
+export class ExchangeEventHandler extends BaseEventHandler<ExchangeEventEntity> {
+    public convertLogEntryToEventEntity(logEntry: LogEntry): ExchangeEventEntity {
+        const decodedLogEntry = decodeLogEntry<ExchangeEventArgs>(this._abi, logEntry);
+        return _convertToEntity(decodedLogEntry);
+    }
 }
 
-// Decodes a LogEntry into a LogWithDecodedArgs
-// tslint:disable-next-line:completed-docs
-export const _decodeLogEntry = R.curry((contractAbi: AbiDefinition[], log: LogEntry): LogWithDecodedArgs<
-    ExchangeEventArgs
-> => {
-    const abiDecoder = new AbiDecoder([contractAbi]);
-    const logWithDecodedArgs = abiDecoder.tryToDecodeLogOrNoop(log);
-    // tslint:disable-next-line:no-unnecessary-type-assertion
-    return logWithDecodedArgs as LogWithDecodedArgs<ExchangeEventArgs>;
-});
-
 export function _convertToEntity(eventLog: LogWithDecodedArgs<ExchangeEventArgs>): ExchangeEventEntity {
     switch (eventLog.event) {
         case 'Fill':
             return _convertToExchangeFillEvent(eventLog as LogWithDecodedArgs<ExchangeFillEventArgs>);
         default:
-            throw new Error('unexpected eventLog.event type: ' + eventLog.event);
+            return new ExchangeFillEvent();
+        // throw new Error('unexpected eventLog.event type: ' + eventLog.event);
     }
 }
 
@@ -116,21 +74,3 @@ function filterEventLogs(
 ): Array<LogWithDecodedArgs<ExchangeEventArgs>> {
     return R.filter(eventLog => eventLog.event === 'Fill', eventLogs);
 }
-
-/**
- * Parses and abi-decodes the raw events response from etherscan.io.
- * @param contractAbi The ABI for the contract that the events where emited from.
- * @param rawEventsResponse The raw events response from etherescan.io.
- * @returns Parsed and decoded event entities, ready to be saved to database.
- */
-export function parseRawEventsResponse(
-    contractAbi: AbiDefinition[],
-    rawEventsResponse: EventsResponse,
-): ExchangeEventEntity[] {
-    return R.pipe(
-        R.map(_convertResponseToLogEntry),
-        R.map(_decodeLogEntry(contractAbi)),
-        filterEventLogs,
-        R.map(_convertToEntity),
-    )(rawEventsResponse.result);
-}
diff --git a/packages/pipeline/src/data_types/events/event_utils.ts b/packages/pipeline/src/data_types/events/event_utils.ts
new file mode 100644
index 000000000..6be964807
--- /dev/null
+++ b/packages/pipeline/src/data_types/events/event_utils.ts
@@ -0,0 +1,35 @@
+import { AbiDecoder } from '@0xproject/utils';
+import { AbiDefinition, LogEntry, LogWithDecodedArgs } from 'ethereum-types';
+
+import { EventsResponseResult } from '../../data_sources/etherscan';
+
+const hexRadix = 16;
+
+function hexToInt(hex: string): number {
+    return parseInt(hex.replace('0x', ''), hexRadix);
+}
+
+// Converts a raw event response to a LogEntry
+export function convertResponseToLogEntry(result: EventsResponseResult): LogEntry {
+    return {
+        logIndex: hexToInt(result.logIndex),
+        transactionIndex: hexToInt(result.transactionIndex),
+        transactionHash: result.transactionHash,
+        blockHash: '',
+        blockNumber: hexToInt(result.blockNumber),
+        address: result.address,
+        data: result.data,
+        topics: result.topics,
+    };
+}
+
+// Decodes a LogEntry into a LogWithDecodedArgs
+export function decodeLogEntry<EventArgsType>(
+    contractAbi: AbiDefinition[],
+    log: LogEntry,
+): LogWithDecodedArgs<EventArgsType> {
+    const abiDecoder = new AbiDecoder([contractAbi]);
+    const logWithDecodedArgs = abiDecoder.tryToDecodeLogOrNoop<EventArgsType>(log);
+    // tslint:disable-next-line:no-unnecessary-type-assertion
+    return logWithDecodedArgs as LogWithDecodedArgs<EventArgsType>;
+}
diff --git a/packages/pipeline/src/entities/ExchangeFillEvent.ts b/packages/pipeline/src/entities/ExchangeFillEvent.ts
index 1716c60d1..a7e817240 100644
--- a/packages/pipeline/src/entities/ExchangeFillEvent.ts
+++ b/packages/pipeline/src/entities/ExchangeFillEvent.ts
@@ -1,9 +1,9 @@
-import { Column, Entity, PrimaryColumn } from 'typeorm';
+import { BaseEntity, Column, Entity, PrimaryColumn } from 'typeorm';
 
 export type ExchangeFillEventAssetType = 'erc20' | 'erc721';
 
 @Entity()
-export class ExchangeFillEvent {
+export class ExchangeFillEvent extends BaseEntity {
     @PrimaryColumn() public logIndex!: number;
 
     @Column() public address!: string;
diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts
index db26343e0..e71a6ae4c 100644
--- a/packages/pipeline/src/index.ts
+++ b/packages/pipeline/src/index.ts
@@ -1,26 +1,32 @@
-import { ExchangeFillEventArgs } from '@0xproject/contract-wrappers';
-import { assetDataUtils } from '@0xproject/order-utils';
-import { LogWithDecodedArgs } from 'ethereum-types';
 import 'reflect-metadata';
 import { createConnection } from 'typeorm';
 
 import { artifacts } from './artifacts';
-import { Etherscan } from './data-sources/etherscan';
+import { Etherscan } from './data_sources/etherscan';
 import { ExchangeFillEvent } from './entities/ExchangeFillEvent';
 import { config } from './ormconfig';
 
+import { ExchangeEventHandler } from './data_types/events/event_handlers/exchange_event_handler';
+
 const etherscan = new Etherscan(process.env.ETHERSCAN_API_KEY as string);
+const EXCHANGE_ADDRESS = '0x4f833a24e1f95d70f028921e27040ca56e09ab0b';
 
 (async () => {
     const connection = await createConnection(config);
     const repository = connection.getRepository(ExchangeFillEvent);
     console.log(`found ${await repository.count()} existing fill events`);
-    const events = await etherscan.getContractEventsAsync(
-        '0x4f833a24e1f95d70f028921e27040ca56e09ab0b',
+    const exchangeEventHandler = new ExchangeEventHandler(
         artifacts.Exchange.compilerOutput.abi,
+        EXCHANGE_ADDRESS,
+        etherscan,
     );
+    const events = await exchangeEventHandler.getEventsAsync();
+    console.log(JSON.stringify(events, null, 2));
     for (const event of events) {
-        await repository.save(event);
+        // TODO(albrow): remove this check once we can parse all Exchange events
+        if (event.address != null) {
+            await event.save();
+        }
     }
     console.log(`now ${await repository.count()} total fill events`);
 })();