Update script to work with existing v1 pipeline data

11 files changed, 155 insertions, 40 deletions

diff --git a/packages/pipeline/package.json b/packages/pipeline/package.json
index 0071fab2c..47033bf33 100644
--- a/packages/pipeline/package.json
+++ b/packages/pipeline/package.json
@@ -25,8 +25,8 @@
     },
     "license": "Apache-2.0",
     "devDependencies": {
-        "@types/ramda": "^0.25.38",
         "@0x/tslint-config": "^1.0.9",
+        "@types/ramda": "^0.25.38",
         "chai": "^4.1.2",
         "chai-as-promised": "^7.1.1",
         "chai-bignumber": "^2.0.2",
@@ -36,10 +36,10 @@
         "typescript": "3.0.1"
     },
     "dependencies": {
-        "@0x/dev-utils": "^1.0.13",
-        "@0x/contract-artifacts": "^1.0.1",
         "@0x/connect": "^3.0.2",
+        "@0x/contract-artifacts": "^1.0.1",
         "@0x/contract-wrappers": "^3.0.0",
+        "@0x/dev-utils": "^1.0.13",
         "@0x/order-utils": "^2.0.0",
         "@0x/subproviders": "^2.1.0",
         "@0x/types": "^1.2.0",
@@ -47,6 +47,7 @@
         "@0x/web3-wrapper": "^3.1.0",
         "axios": "^0.18.0",
         "ethereum-types": "^1.0.6",
+        "pg": "^7.5.0",
         "ramda": "^0.25.0",
         "reflect-metadata": "^0.1.12",
         "sqlite3": "^4.0.2",
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 46f115705..5ce8381cd 100644
--- a/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
+++ b/packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts
@@ -29,7 +29,10 @@ export class ExchangeEventsSource {
         let events: Array<LogWithDecodedArgs<ExchangeFillEventArgs>> = [];
         for (let currFromBlock = fromBlock; currFromBlock <= calculatedToBlock; currFromBlock += NUM_BLOCKS_PER_QUERY) {
             events = events.concat(
-                await this._getFillEventsForRangeAsync(currFromBlock, currFromBlock + NUM_BLOCKS_PER_QUERY - 1),
+                await this._getFillEventsForRangeAsync(
+                    currFromBlock,
+                    Math.min(currFromBlock + NUM_BLOCKS_PER_QUERY - 1, calculatedToBlock),
+                ),
             );
         }
         return events;
diff --git a/packages/pipeline/src/entities/ExchangeCancelEvent.ts b/packages/pipeline/src/entities/ExchangeCancelEvent.ts
index 7010ab9f2..698b9e2ec 100644
--- a/packages/pipeline/src/entities/ExchangeCancelEvent.ts
+++ b/packages/pipeline/src/entities/ExchangeCancelEvent.ts
@@ -1,15 +1,16 @@
-import { BaseEntity, Column, Entity, PrimaryColumn } from 'typeorm';
+import { Column, Entity, PrimaryColumn } from 'typeorm';
 
 import { AssetType } from '../types';
 
 @Entity()
-export class ExchangeCancelEvent extends BaseEntity {
+export class ExchangeCancelEvent {
     @PrimaryColumn() public contractAddress!: string;
     @PrimaryColumn() public logIndex!: number;
     @PrimaryColumn() public blockNumber!: number;
 
     @Column() public rawData!: string;
 
+    // TODO(albrow): Include transaction hash
     @Column() public makerAddress!: string;
     @Column({ nullable: true, type: String })
     public takerAddress!: string;
diff --git a/packages/pipeline/src/entities/ExchangeCancelUpToEvent.ts b/packages/pipeline/src/entities/ExchangeCancelUpToEvent.ts
index 03a02b069..3ca75ccf7 100644
--- a/packages/pipeline/src/entities/ExchangeCancelUpToEvent.ts
+++ b/packages/pipeline/src/entities/ExchangeCancelUpToEvent.ts
@@ -1,11 +1,12 @@
-import { BaseEntity, Column, Entity, PrimaryColumn } from 'typeorm';
+import { Column, Entity, PrimaryColumn } from 'typeorm';
 
 @Entity()
-export class ExchangeCancelUpToEvent extends BaseEntity {
+export class ExchangeCancelUpToEvent {
     @PrimaryColumn() public contractAddress!: string;
     @PrimaryColumn() public logIndex!: number;
     @PrimaryColumn() public blockNumber!: number;
 
+    // TODO(albrow): Include transaction hash
     @Column() public rawData!: string;
 
     @Column() public makerAddress!: string;
diff --git a/packages/pipeline/src/entities/ExchangeFillEvent.ts b/packages/pipeline/src/entities/ExchangeFillEvent.ts
index 5eafa7449..6e549af93 100644
--- a/packages/pipeline/src/entities/ExchangeFillEvent.ts
+++ b/packages/pipeline/src/entities/ExchangeFillEvent.ts
@@ -1,18 +1,19 @@
-import { BaseEntity, Column, Entity, PrimaryColumn } from 'typeorm';
+import { Column, Entity, PrimaryColumn } from 'typeorm';
 
 import { AssetType } from '../types';
 
 @Entity()
-export class ExchangeFillEvent extends BaseEntity {
+export class ExchangeFillEvent {
     @PrimaryColumn() public contractAddress!: string;
     @PrimaryColumn() public logIndex!: number;
     @PrimaryColumn() public blockNumber!: number;
 
     @Column() public rawData!: string;
 
+    @Column() public transactionHash!: string;
     @Column() public makerAddress!: string;
     @Column() public takerAddress!: string;
-    @Column() public feeRecepientAddress!: string;
+    @Column() public feeRecipientAddress!: string;
     @Column() public senderAddress!: string;
     @Column() public makerAssetFilledAmount!: string;
     @Column() public takerAssetFilledAmount!: string;
diff --git a/packages/pipeline/src/entities/SraOrder.ts b/packages/pipeline/src/entities/SraOrder.ts
index e4987df57..a22f7c4e5 100644
--- a/packages/pipeline/src/entities/SraOrder.ts
+++ b/packages/pipeline/src/entities/SraOrder.ts
@@ -1,9 +1,9 @@
-import { BaseEntity, Column, Entity, PrimaryColumn } from 'typeorm';
+import { Column, Entity, PrimaryColumn } from 'typeorm';
 
 import { AssetType } from '../types';
 
 @Entity()
-export class SraOrder extends BaseEntity {
+export class SraOrder {
     @PrimaryColumn() public exchangeAddress!: string;
     @PrimaryColumn() public orderHashHex!: string;
 
diff --git a/packages/pipeline/src/index.ts b/packages/pipeline/src/index.ts
index d4bca65c5..58646fc56 100644
--- a/packages/pipeline/src/index.ts
+++ b/packages/pipeline/src/index.ts
@@ -1,50 +1,78 @@
-import { HttpClient } from '@0x/connect';
 import { web3Factory } from '@0x/dev-utils';
 import 'reflect-metadata';
 import { Connection, createConnection } from 'typeorm';
 
 import { ExchangeEventsSource } from './data_sources/contract-wrappers/exchange_events';
-import { SraOrder } from './entities/SraOrder';
-import { config } from './ormconfig';
+import { deployConfig } from './ormconfig';
 import { parseExchangeEvents } from './parsers/events';
-import { parseSraOrders } from './parsers/sra_orders';
+import { ExchangeFillEvent } from './entities/ExchangeFillEvent';
 
 let connection: Connection;
 
 (async () => {
-    connection = await createConnection(config);
+    connection = await createConnection(deployConfig);
     await getExchangeEventsAsync();
-    // await getSraOrdersAsync();
+    await mergeExchangeEventsAsync();
+    console.log('Exiting process');
+    process.exit(0);
 })();
 
 // TODO(albrow): Separately: Errors do not appear to be handled correctly. If you use the
 // wrong rpcUrl it just returns early with no error.
 async function getExchangeEventsAsync(): Promise<void> {
+    console.log('Getting event logs...');
     const provider = web3Factory.getRpcProvider({
         rpcUrl: 'https://mainnet.infura.io',
     });
+    const eventsRepository = connection.getRepository(ExchangeFillEvent);
     const exchangeEvents = new ExchangeEventsSource(provider, 1);
     const eventLogs = await exchangeEvents.getFillEventsAsync();
+    console.log('Parsing events...');
     const events = parseExchangeEvents(eventLogs);
-    console.log('Got events: ' + events.length);
-    for (const event of events) {
-        await event.save();
-    }
+    console.log(`Retrieved and parsed ${events.length} total events.`);
+    console.log('Saving events...');
+    eventsRepository.save(events);
     console.log('Saved events.');
-    console.log('Exiting process');
-    process.exit(0);
 }
 
-async function getSraOrdersAsync(): Promise<void> {
-    const orderRepository = connection.getRepository(SraOrder);
-    console.log(`found ${await orderRepository.count()} existing orders`);
-    const sraUrl = 'https://api.radarrelay.com/0x/v2';
-    const connect = new HttpClient(sraUrl);
-    const rawOrders = await connect.getOrdersAsync();
-    const orders = parseSraOrders(rawOrders);
-    for (const order of orders) {
-        order.sourceUrl = sraUrl;
-        await order.save();
-    }
-    console.log(`now there are ${await orderRepository.count()} total orders`);
+const insertEventsRawQuery = `INSERT INTO events_raw (
+    event_type,
+    error_id,
+    order_hash,
+    maker,
+    maker_amount,
+    maker_fee,
+    maker_token,
+    taker,
+    taker_amount,
+    taker_fee,
+    taker_token,
+    txn_hash,
+    fee_recipient,
+    block_number,
+    log_index
+)
+(
+    SELECT
+        'LogFill',
+        null,
+        "orderHash",
+        "makerAddress",
+        "makerAssetFilledAmount"::numeric(78),
+        "makerFeePaid"::numeric(78),
+        "makerTokenAddress",
+        "takerAddress",
+        "takerAssetFilledAmount"::numeric(78),
+        "takerFeePaid"::numeric(78),
+        "takerTokenAddress",
+        "transactionHash",
+        "feeRecipientAddress",
+        "blockNumber",
+        "logIndex"
+    FROM exchange_fill_event
+) ON CONFLICT (order_hash, txn_hash, log_index) DO NOTHING`;
+
+async function mergeExchangeEventsAsync(): Promise<void> {
+    console.log('Merging results into events_raw...');
+    await connection.query(insertEventsRawQuery);
 }
diff --git a/packages/pipeline/src/ormconfig.ts b/packages/pipeline/src/ormconfig.ts
index 48e316078..eaafeb7d5 100644
--- a/packages/pipeline/src/ormconfig.ts
+++ b/packages/pipeline/src/ormconfig.ts
@@ -1,6 +1,6 @@
 import { ConnectionOptions } from 'typeorm';
 
-export const config: ConnectionOptions = {
+export const testConfig: ConnectionOptions = {
     type: 'sqlite',
     database: 'database.sqlite',
     synchronize: true,
@@ -12,3 +12,16 @@ export const config: ConnectionOptions = {
         migrationsDir: 'lib/src/migrations',
     },
 };
+
+export const deployConfig: ConnectionOptions = {
+    type: 'postgres',
+    url: process.env.ZEROEX_DATA_PIPELINE_DB_URL,
+    synchronize: true,
+    logging: false,
+    entities: ['./lib/src/entities/**/*.js'],
+    migrations: ['./lib/src/migrations/**/*.js'],
+    cli: {
+        entitiesDir: 'lib/src/entities',
+        migrationsDir: 'lib/src/migrations',
+    },
+};
diff --git a/packages/pipeline/src/parsers/events/index.ts b/packages/pipeline/src/parsers/events/index.ts
index b9b4d02cb..abfb7ddb6 100644
--- a/packages/pipeline/src/parsers/events/index.ts
+++ b/packages/pipeline/src/parsers/events/index.ts
@@ -43,9 +43,10 @@ export function _convertToExchangeFillEvent(eventLog: LogWithDecodedArgs<Exchang
     exchangeFillEvent.blockNumber = eventLog.blockNumber as number;
     exchangeFillEvent.logIndex = eventLog.logIndex as number;
     exchangeFillEvent.rawData = eventLog.data as string;
+    exchangeFillEvent.transactionHash = eventLog.transactionHash;
     exchangeFillEvent.makerAddress = eventLog.args.makerAddress.toString();
     exchangeFillEvent.takerAddress = eventLog.args.takerAddress.toString();
-    exchangeFillEvent.feeRecepientAddress = eventLog.args.feeRecipientAddress;
+    exchangeFillEvent.feeRecipientAddress = eventLog.args.feeRecipientAddress;
     exchangeFillEvent.senderAddress = eventLog.args.senderAddress;
     exchangeFillEvent.makerAssetFilledAmount = eventLog.args.makerAssetFilledAmount.toString();
     exchangeFillEvent.takerAssetFilledAmount = eventLog.args.takerAssetFilledAmount.toString();
diff --git a/packages/pipeline/test/parsers/events/index_test.ts b/packages/pipeline/test/parsers/events/index_test.ts
index 0730e4013..451988f8e 100644
--- a/packages/pipeline/test/parsers/events/index_test.ts
+++ b/packages/pipeline/test/parsers/events/index_test.ts
@@ -51,9 +51,10 @@ describe('exchange_events', () => {
             expected.logIndex = 102;
             expected.rawData =
                 '0x000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428000000000000000000000000f6da68519f78b0d0bc93c701e86affcb75c92428000000000000000000000000000000000000000000000000002386f26fc10000000000000000000000000000000000000000000000000000016345785d8a000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000010000000000000000000000000000000000000000000000000000000000000001600000000000000000000000000000000000000000000000000000000000000024f47261b0000000000000000000000000c02aaa39b223fe8d0a0e5c4f27ead9083c756cc2000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000024f47261b0000000000000000000000000e41d2489571d322189246dafa5ebde1f4699f49800000000000000000000000000000000000000000000000000000000';
+            expected.transactionHash = '0x6dd106d002873746072fc5e496dd0fb2541b68c77bcf9184ae19a42fd33657fe';
             expected.makerAddress = '0xf6da68519f78b0d0bc93c701e86affcb75c92428';
             expected.takerAddress = '0xf6da68519f78b0d0bc93c701e86affcb75c92428';
-            expected.feeRecepientAddress = '0xc370d2a5920344aa6b7d8d11250e3e861434cbdd';
+            expected.feeRecipientAddress = '0xc370d2a5920344aa6b7d8d11250e3e861434cbdd';
             expected.senderAddress = '0xf6da68519f78b0d0bc93c701e86affcb75c92428';
             expected.makerAssetFilledAmount = '10000000000000000';
             expected.takerAssetFilledAmount = '100000000000000000';
diff --git a/yarn.lock b/yarn.lock
index a73a2b8aa..517da9451 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -3614,6 +3614,10 @@ buffer-to-arraybuffer@^0.0.5:
   version "0.0.5"
   resolved "https://registry.yarnpkg.com/buffer-to-arraybuffer/-/buffer-to-arraybuffer-0.0.5.tgz#6064a40fa76eb43c723aba9ef8f6e1216d10511a"
 
+buffer-writer@1.0.1:
+  version "1.0.1"
+  resolved "https://registry.yarnpkg.com/buffer-writer/-/buffer-writer-1.0.1.tgz#22a936901e3029afcd7547eb4487ceb697a3bf08"
+
 buffer-xor@^1.0.3:
   version "1.0.3"
   resolved "https://registry.yarnpkg.com/buffer-xor/-/buffer-xor-1.0.3.tgz#26e61ed1422fb70dd42e6e36729ed51d855fe8d9"
@@ -11663,6 +11667,10 @@ package-json@^4.0.0, package-json@^4.0.1:
     registry-url "^3.0.3"
     semver "^5.1.0"
 
+packet-reader@0.3.1:
+  version "0.3.1"
+  resolved "https://registry.yarnpkg.com/packet-reader/-/packet-reader-0.3.1.tgz#cd62e60af8d7fea8a705ec4ff990871c46871f27"
+
 pacote@^9.1.0:
   version "9.1.0"
   resolved "https://registry.yarnpkg.com/pacote/-/pacote-9.1.0.tgz#59810859bbd72984dcb267269259375d32f391e5"
@@ -11935,6 +11943,41 @@ performance-now@^2.1.0:
   version "2.1.0"
   resolved "https://registry.yarnpkg.com/performance-now/-/performance-now-2.1.0.tgz#6309f4e0e5fa913ec1c69307ae364b4b377c9e7b"
 
+pg-connection-string@0.1.3:
+  version "0.1.3"
+  resolved "https://registry.yarnpkg.com/pg-connection-string/-/pg-connection-string-0.1.3.tgz#da1847b20940e42ee1492beaf65d49d91b245df7"
+
+pg-pool@~2.0.3:
+  version "2.0.3"
+  resolved "https://registry.yarnpkg.com/pg-pool/-/pg-pool-2.0.3.tgz#c022032c8949f312a4f91fb6409ce04076be3257"
+
+pg-types@~1.12.1:
+  version "1.12.1"
+  resolved "https://registry.yarnpkg.com/pg-types/-/pg-types-1.12.1.tgz#d64087e3903b58ffaad279e7595c52208a14c3d2"
+  dependencies:
+    postgres-array "~1.0.0"
+    postgres-bytea "~1.0.0"
+    postgres-date "~1.0.0"
+    postgres-interval "^1.1.0"
+
+pg@^7.5.0:
+  version "7.5.0"
+  resolved "https://registry.yarnpkg.com/pg/-/pg-7.5.0.tgz#c2853bef2fcb91424ba2f649fd951ce866a84760"
+  dependencies:
+    buffer-writer "1.0.1"
+    packet-reader "0.3.1"
+    pg-connection-string "0.1.3"
+    pg-pool "~2.0.3"
+    pg-types "~1.12.1"
+    pgpass "1.x"
+    semver "4.3.2"
+
+pgpass@1.x:
+  version "1.0.2"
+  resolved "https://registry.yarnpkg.com/pgpass/-/pgpass-1.0.2.tgz#2a7bb41b6065b67907e91da1b07c1847c877b306"
+  dependencies:
+    split "^1.0.0"
+
 pify@^2.0.0, pify@^2.2.0, pify@^2.3.0:
   version "2.3.0"
   resolved "https://registry.yarnpkg.com/pify/-/pify-2.3.0.tgz#ed141a6ac043a849ea588498e7dca8b15330e90c"
@@ -12263,6 +12306,24 @@ postcss@^6.0.1:
     source-map "^0.6.1"
     supports-color "^5.3.0"
 
+postgres-array@~1.0.0:
+  version "1.0.3"
+  resolved "https://registry.yarnpkg.com/postgres-array/-/postgres-array-1.0.3.tgz#c561fc3b266b21451fc6555384f4986d78ec80f5"
+
+postgres-bytea@~1.0.0:
+  version "1.0.0"
+  resolved "https://registry.yarnpkg.com/postgres-bytea/-/postgres-bytea-1.0.0.tgz#027b533c0aa890e26d172d47cf9ccecc521acd35"
+
+postgres-date@~1.0.0:
+  version "1.0.3"
+  resolved "https://registry.yarnpkg.com/postgres-date/-/postgres-date-1.0.3.tgz#e2d89702efdb258ff9d9cee0fe91bd06975257a8"
+
+postgres-interval@^1.1.0:
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/postgres-interval/-/postgres-interval-1.1.2.tgz#bf71ff902635f21cb241a013fc421d81d1db15a9"
+  dependencies:
+    xtend "^4.0.0"
+
 prebuild-install@^2.2.2:
   version "2.5.3"
   resolved "https://registry.yarnpkg.com/prebuild-install/-/prebuild-install-2.5.3.tgz#9f65f242782d370296353710e9bc843490c19f69"
@@ -14048,6 +14109,10 @@ semver-sort@0.0.4:
   version "5.5.0"
   resolved "https://registry.yarnpkg.com/semver/-/semver-5.5.0.tgz#dc4bbc7a6ca9d916dee5d43516f0092b58f7b8ab"
 
+semver@4.3.2:
+  version "4.3.2"
+  resolved "http://registry.npmjs.org/semver/-/semver-4.3.2.tgz#c7a07158a80bedd052355b770d82d6640f803be7"
+
 semver@^4.1.0:
   version "4.3.6"
   resolved "https://registry.yarnpkg.com/semver/-/semver-4.3.6.tgz#300bc6e0e86374f7ba61068b5b1ecd57fc6532da"