Merge branch 'development' into feature/instant/usd-eth-toggle

5 files changed, 322 insertions, 51 deletions

diff --git a/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts b/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts
new file mode 100644
index 000000000..e0098122f
--- /dev/null
+++ b/packages/pipeline/src/data_sources/contract-wrappers/erc20_events.ts
@@ -0,0 +1,45 @@
+import {
+    ContractWrappers,
+    ERC20TokenApprovalEventArgs,
+    ERC20TokenEvents,
+    ERC20TokenWrapper,
+} from '@0x/contract-wrappers';
+import { Web3ProviderEngine } from '@0x/subproviders';
+import { LogWithDecodedArgs } from 'ethereum-types';
+
+import { GetEventsFunc, getEventsWithPaginationAsync } from './utils';
+
+export class ERC20EventsSource {
+    private readonly _erc20Wrapper: ERC20TokenWrapper;
+    private readonly _tokenAddress: string;
+    constructor(provider: Web3ProviderEngine, networkId: number, tokenAddress: string) {
+        const contractWrappers = new ContractWrappers(provider, { networkId });
+        this._erc20Wrapper = contractWrappers.erc20Token;
+        this._tokenAddress = tokenAddress;
+    }
+
+    public async getApprovalEventsAsync(
+        startBlock: number,
+        endBlock: number,
+    ): Promise<Array<LogWithDecodedArgs<ERC20TokenApprovalEventArgs>>> {
+        return getEventsWithPaginationAsync(
+            this._getApprovalEventsForRangeAsync.bind(this) as GetEventsFunc<ERC20TokenApprovalEventArgs>,
+            startBlock,
+            endBlock,
+        );
+    }
+
+    // Gets all approval events of for a specific sub-range. This getter
+    // function will be called during each step of pagination.
+    private async _getApprovalEventsForRangeAsync(
+        fromBlock: number,
+        toBlock: number,
+    ): Promise<Array<LogWithDecodedArgs<ERC20TokenApprovalEventArgs>>> {
+        return this._erc20Wrapper.getLogsAsync<ERC20TokenApprovalEventArgs>(
+            this._tokenAddress,
+            ERC20TokenEvents.Approval,
+            { fromBlock, toBlock },
+            {},
+        );
+    }
+}
diff --git a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
index 1717eb8b3..58691e2ab 100644
--- a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
+++ b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
@@ -8,78 +8,52 @@ import {
     ExchangeWrapper,
 } from '@0x/contract-wrappers';
 import { Web3ProviderEngine } from '@0x/subproviders';
-import { Web3Wrapper } from '@0x/web3-wrapper';
 import { LogWithDecodedArgs } from 'ethereum-types';
 
-import { EXCHANGE_START_BLOCK } from '../../utils';
-
-const BLOCK_FINALITY_THRESHOLD = 10; // When to consider blocks as final. Used to compute default toBlock.
-const NUM_BLOCKS_PER_QUERY = 20000; // Number of blocks to query for events at a time.
+import { GetEventsFunc, getEventsWithPaginationAsync } from './utils';
 
 export class ExchangeEventsSource {
     private readonly _exchangeWrapper: ExchangeWrapper;
-    private readonly _web3Wrapper: Web3Wrapper;
     constructor(provider: Web3ProviderEngine, networkId: number) {
-        this._web3Wrapper = new Web3Wrapper(provider);
         const contractWrappers = new ContractWrappers(provider, { networkId });
         this._exchangeWrapper = contractWrappers.exchange;
     }
 
     public async getFillEventsAsync(
-        fromBlock?: number,
-        toBlock?: number,
+        startBlock: number,
+        endBlock: number,
     ): Promise<Array<LogWithDecodedArgs<ExchangeFillEventArgs>>> {
-        return this._getEventsAsync<ExchangeFillEventArgs>(ExchangeEvents.Fill, fromBlock, toBlock);
+        const getFillEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeFillEventArgs>(ExchangeEvents.Fill);
+        return getEventsWithPaginationAsync(getFillEventsForRangeAsync, startBlock, endBlock);
     }
 
     public async getCancelEventsAsync(
-        fromBlock?: number,
-        toBlock?: number,
+        startBlock: number,
+        endBlock: number,
     ): Promise<Array<LogWithDecodedArgs<ExchangeCancelEventArgs>>> {
-        return this._getEventsAsync<ExchangeCancelEventArgs>(ExchangeEvents.Cancel, fromBlock, toBlock);
+        const getCancelEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeCancelEventArgs>(
+            ExchangeEvents.Cancel,
+        );
+        return getEventsWithPaginationAsync(getCancelEventsForRangeAsync, startBlock, endBlock);
     }
 
     public async getCancelUpToEventsAsync(
-        fromBlock?: number,
-        toBlock?: number,
+        startBlock: number,
+        endBlock: number,
     ): Promise<Array<LogWithDecodedArgs<ExchangeCancelUpToEventArgs>>> {
-        return this._getEventsAsync<ExchangeCancelUpToEventArgs>(ExchangeEvents.CancelUpTo, fromBlock, toBlock);
-    }
-
-    private async _getEventsAsync<ArgsType extends ExchangeEventArgs>(
-        eventName: ExchangeEvents,
-        fromBlock: number = EXCHANGE_START_BLOCK,
-        toBlock?: number,
-    ): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
-        const calculatedToBlock =
-            toBlock === undefined
-                ? (await this._web3Wrapper.getBlockNumberAsync()) - BLOCK_FINALITY_THRESHOLD
-                : toBlock;
-        let events: Array<LogWithDecodedArgs<ArgsType>> = [];
-        for (let currFromBlock = fromBlock; currFromBlock <= calculatedToBlock; currFromBlock += NUM_BLOCKS_PER_QUERY) {
-            events = events.concat(
-                await this._getEventsForRangeAsync<ArgsType>(
-                    eventName,
-                    currFromBlock,
-                    Math.min(currFromBlock + NUM_BLOCKS_PER_QUERY - 1, calculatedToBlock),
-                ),
-            );
-        }
-        return events;
+        const getCancelUpToEventsForRangeAsync = this._makeGetterFuncForEventType<ExchangeCancelUpToEventArgs>(
+            ExchangeEvents.CancelUpTo,
+        );
+        return getEventsWithPaginationAsync(getCancelUpToEventsForRangeAsync, startBlock, endBlock);
     }
 
-    private async _getEventsForRangeAsync<ArgsType extends ExchangeEventArgs>(
-        eventName: ExchangeEvents,
-        fromBlock: number,
-        toBlock: number,
-    ): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
-        return this._exchangeWrapper.getLogsAsync<ArgsType>(
-            eventName,
-            {
-                fromBlock,
-                toBlock,
-            },
-            {},
-        );
+    // Returns a getter function which gets all events of a specific type for a
+    // specific sub-range. This getter function will be called during each step
+    // of pagination.
+    private _makeGetterFuncForEventType<ArgsType extends ExchangeEventArgs>(
+        eventType: ExchangeEvents,
+    ): GetEventsFunc<ArgsType> {
+        return async (fromBlock: number, toBlock: number) =>
+            this._exchangeWrapper.getLogsAsync<ArgsType>(eventType, { fromBlock, toBlock }, {});
     }
 }
diff --git a/packages/pipeline/src/data_sources/contract-wrappers/utils.ts b/packages/pipeline/src/data_sources/contract-wrappers/utils.ts
new file mode 100644
index 000000000..67660a37e
--- /dev/null
+++ b/packages/pipeline/src/data_sources/contract-wrappers/utils.ts
@@ -0,0 +1,67 @@
+import { DecodedLogArgs, LogWithDecodedArgs } from 'ethereum-types';
+
+const NUM_BLOCKS_PER_QUERY = 10000; // Number of blocks to query for events at a time.
+const NUM_RETRIES = 3; // Number of retries if a request fails or times out.
+
+export type GetEventsFunc<ArgsType extends DecodedLogArgs> = (
+    fromBlock: number,
+    toBlock: number,
+) => Promise<Array<LogWithDecodedArgs<ArgsType>>>;
+
+/**
+ * Gets all events between the given startBlock and endBlock by querying for
+ * NUM_BLOCKS_PER_QUERY at a time. Accepts a getter function in order to
+ * maximize code re-use and allow for getting different types of events for
+ * different contracts. If the getter function throws with a retryable error,
+ * it will automatically be retried up to NUM_RETRIES times.
+ * @param getEventsAsync A getter function which will be called for each step during pagination.
+ * @param startBlock The start of the entire block range to get events for.
+ * @param endBlock The end of the entire block range to get events for.
+ */
+export async function getEventsWithPaginationAsync<ArgsType extends DecodedLogArgs>(
+    getEventsAsync: GetEventsFunc<ArgsType>,
+    startBlock: number,
+    endBlock: number,
+): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
+    let events: Array<LogWithDecodedArgs<ArgsType>> = [];
+    for (let fromBlock = startBlock; fromBlock <= endBlock; fromBlock += NUM_BLOCKS_PER_QUERY) {
+        const toBlock = Math.min(fromBlock + NUM_BLOCKS_PER_QUERY - 1, endBlock);
+        const eventsInRange = await _getEventsWithRetriesAsync(getEventsAsync, NUM_RETRIES, fromBlock, toBlock);
+        events = events.concat(eventsInRange);
+    }
+    return events;
+}
+
+/**
+ * Calls the getEventsAsync function and retries up to numRetries times if it
+ * throws with an error that is considered retryable.
+ * @param getEventsAsync a function that will be called on each iteration.
+ * @param numRetries the maximum number times to retry getEventsAsync if it fails with a retryable error.
+ * @param fromBlock the start of the sub-range of blocks we are getting events for.
+ * @param toBlock the end of the sub-range of blocks we are getting events for.
+ */
+export async function _getEventsWithRetriesAsync<ArgsType extends DecodedLogArgs>(
+    getEventsAsync: GetEventsFunc<ArgsType>,
+    numRetries: number,
+    fromBlock: number,
+    toBlock: number,
+): Promise<Array<LogWithDecodedArgs<ArgsType>>> {
+    let eventsInRange: Array<LogWithDecodedArgs<ArgsType>> = [];
+    for (let i = 0; i <= numRetries; i++) {
+        try {
+            eventsInRange = await getEventsAsync(fromBlock, toBlock);
+        } catch (err) {
+            if (isErrorRetryable(err) && i < numRetries) {
+                continue;
+            } else {
+                throw err;
+            }
+        }
+        break;
+    }
+    return eventsInRange;
+}
+
+function isErrorRetryable(err: Error): boolean {
+    return err.message.includes('network timeout');
+}
diff --git a/packages/pipeline/src/data_sources/idex/index.ts b/packages/pipeline/src/data_sources/idex/index.ts
new file mode 100644
index 000000000..c1e53c08d
--- /dev/null
+++ b/packages/pipeline/src/data_sources/idex/index.ts
@@ -0,0 +1,82 @@
+import { fetchAsync } from '@0x/utils';
+
+const IDEX_BASE_URL = 'https://api.idex.market';
+const MARKETS_URL = `${IDEX_BASE_URL}/returnTicker`;
+const ORDERBOOK_URL = `${IDEX_BASE_URL}/returnOrderBook`;
+const MAX_ORDER_COUNT = 100; // Maximum based on https://github.com/AuroraDAO/idex-api-docs#returnorderbook
+export const IDEX_SOURCE = 'idex';
+
+export interface IdexMarketsResponse {
+    [marketName: string]: IdexMarket;
+}
+
+export interface IdexMarket {
+    last: string;
+    high: string;
+    low: string;
+    lowestAsk: string;
+    highestBid: string;
+    percentChange: string;
+    baseVolume: string;
+    quoteVolume: string;
+}
+
+export interface IdexOrderbook {
+    asks: IdexOrder[];
+    bids: IdexOrder[];
+}
+
+export interface IdexOrder {
+    price: string;
+    amount: string;
+    total: string;
+    orderHash: string;
+    params: IdexOrderParam;
+}
+
+export interface IdexOrderParam {
+    tokenBuy: string;
+    buySymbol: string;
+    buyPrecision: number;
+    amountBuy: string;
+    tokenSell: string;
+    sellSymbol: string;
+    sellPrecision: number;
+    amountSell: string;
+    expires: number;
+    nonce: number;
+    user: string;
+}
+
+// tslint:disable:prefer-function-over-method
+// ^ Keep consistency with other sources and help logical organization
+export class IdexSource {
+    /**
+     * Call Idex API to find out which markets they are maintaining orderbooks for.
+     */
+    public async getMarketsAsync(): Promise<string[]> {
+        const params = { method: 'POST' };
+        const resp = await fetchAsync(MARKETS_URL, params);
+        const respJson: IdexMarketsResponse = await resp.json();
+        const markets: string[] = Object.keys(respJson);
+        return markets;
+    }
+
+    /**
+     * Retrieve orderbook from Idex API for a given market.
+     * @param marketId String identifying the market we want data for. Eg. 'REP_AUG'
+     */
+    public async getMarketOrderbookAsync(marketId: string): Promise<IdexOrderbook> {
+        const params = {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({
+                market: marketId,
+                count: MAX_ORDER_COUNT,
+            }),
+        };
+        const resp = await fetchAsync(ORDERBOOK_URL, params);
+        const respJson: IdexOrderbook = await resp.json();
+        return respJson;
+    }
+}
diff --git a/packages/pipeline/src/data_sources/oasis/index.ts b/packages/pipeline/src/data_sources/oasis/index.ts
new file mode 100644
index 000000000..3b30e9dfd
--- /dev/null
+++ b/packages/pipeline/src/data_sources/oasis/index.ts
@@ -0,0 +1,103 @@
+import { fetchAsync } from '@0x/utils';
+
+const OASIS_BASE_URL = 'https://data.makerdao.com/v1';
+const OASIS_MARKET_QUERY = `query {
+  oasisMarkets(period: "1 week") {
+    nodes {
+      id
+      base
+      quote
+      buyVol
+      sellVol
+      price
+      high
+      low
+    }
+  }
+}`;
+const OASIS_ORDERBOOK_QUERY = `query ($market: String!) {
+  allOasisOrders(condition: { market: $market }) {
+    totalCount
+    nodes {
+      market
+      offerId
+      price
+      amount
+      act
+    }
+  }
+}`;
+export const OASIS_SOURCE = 'oasis';
+
+export interface OasisMarket {
+    id: string; // market symbol e.g MKRDAI
+    base: string; // base symbol   e.g MKR
+    quote: string; // quote symbol  e.g DAI
+    buyVol: number; // total buy volume (base)
+    sellVol: number; // total sell volume (base)
+    price: number; // volume weighted price (quote)
+    high: number; // max sell price
+    low: number; // min buy price
+}
+
+export interface OasisMarketResponse {
+    data: {
+        oasisMarkets: {
+            nodes: OasisMarket[];
+        };
+    };
+}
+
+export interface OasisOrder {
+    offerId: number; // Offer Id
+    market: string; // Market symbol (base/quote)
+    price: string; // Offer price (quote)
+    amount: string; // Offer amount (base)
+    act: string; // Action (ask|bid)
+}
+
+export interface OasisOrderbookResponse {
+    data: {
+        allOasisOrders: {
+            totalCount: number;
+            nodes: OasisOrder[];
+        };
+    };
+}
+
+// tslint:disable:prefer-function-over-method
+// ^ Keep consistency with other sources and help logical organization
+export class OasisSource {
+    /**
+     * Call Ddex API to find out which markets they are maintaining orderbooks for.
+     */
+    public async getActiveMarketsAsync(): Promise<OasisMarket[]> {
+        const params = {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ query: OASIS_MARKET_QUERY }),
+        };
+        const resp = await fetchAsync(OASIS_BASE_URL, params);
+        const respJson: OasisMarketResponse = await resp.json();
+        const markets = respJson.data.oasisMarkets.nodes;
+        return markets;
+    }
+
+    /**
+     * Retrieve orderbook from Oasis API for a given market.
+     * @param marketId String identifying the market we want data for. Eg. 'REPAUG'.
+     */
+    public async getMarketOrderbookAsync(marketId: string): Promise<OasisOrder[]> {
+        const input = {
+            market: marketId,
+        };
+        const params = {
+            method: 'POST',
+            headers: { 'Content-Type': 'application/json' },
+            body: JSON.stringify({ query: OASIS_ORDERBOOK_QUERY, variables: input }),
+        };
+        const resp = await fetchAsync(OASIS_BASE_URL, params);
+        const respJson: OasisOrderbookResponse = await resp.json();
+        return respJson.data.allOasisOrders.nodes;
+    }
+}