Add better error handling for immediately invoked async functions

4 files changed, 18 insertions, 95 deletions

diff --git a/packages/pipeline/src/entities/block.ts b/packages/pipeline/src/entities/block.ts
index 49e0ef840..5bd51f3d2 100644
--- a/packages/pipeline/src/entities/block.ts
+++ b/packages/pipeline/src/entities/block.ts
@@ -1,9 +1,10 @@
 import { Column, Entity, PrimaryColumn } from 'typeorm';
 
-@Entity()
+@Entity({ name: 'blocks' })
 export class Block {
     @PrimaryColumn() public hash!: string;
     @PrimaryColumn() public number!: number;
 
-    @Column() public unixTimestampSeconds!: number;
+    @Column({ name: 'unix_timestamp_seconds' })
+    public unixTimestampSeconds!: number;
 }
diff --git a/packages/pipeline/src/entities/relayer.ts b/packages/pipeline/src/entities/relayer.ts
index 605355fa8..94b5232a8 100644
--- a/packages/pipeline/src/entities/relayer.ts
+++ b/packages/pipeline/src/entities/relayer.ts
@@ -1,23 +1,23 @@
 import { Column, Entity, PrimaryColumn } from 'typeorm';
 
-@Entity()
+@Entity({ name: 'relayers' })
 export class Relayer {
     @PrimaryColumn() public uuid!: string;
 
     @Column() public name!: string;
     @Column() public url!: string;
-    @Column({ nullable: true, type: String })
+    @Column({ nullable: true, type: String, name: 'sra_http_endpoint' })
     public sraHttpEndpoint!: string | null;
-    @Column({ nullable: true, type: String })
+    @Column({ nullable: true, type: String, name: 'sra_ws_endpoint' })
     public sraWsEndpoint!: string | null;
-    @Column({ nullable: true, type: String })
+    @Column({ nullable: true, type: String, name: 'app_url' })
     public appUrl!: string | null;
 
     // TODO(albrow): Add exchange contract or protocol version?
     // TODO(albrow): Add network ids for addresses?
 
-    @Column({ type: 'varchar', array: true })
+    @Column({ type: 'varchar', array: true, name: 'fee_recipient_addresses' })
     public feeRecipientAddresses!: string[];
-    @Column({ type: 'varchar', array: true })
+    @Column({ type: 'varchar', array: true, name: 'taker_addresses' })
     public takerAddresses!: string[];
 }
diff --git a/packages/pipeline/src/entities/transaction.ts b/packages/pipeline/src/entities/transaction.ts
index 8a68da813..eb2883fda 100644
--- a/packages/pipeline/src/entities/transaction.ts
+++ b/packages/pipeline/src/entities/transaction.ts
@@ -1,13 +1,16 @@
 import { Column, Entity, PrimaryColumn } from 'typeorm';
 
-@Entity()
+@Entity({ name: 'transactions' })
 export class Transaction {
-    @PrimaryColumn() public transactionHash!: string;
-    @PrimaryColumn() public blockHash!: string;
-    @PrimaryColumn() public blockNumber!: number;
+    @PrimaryColumn({ name: 'transaction_hash' })
+    public transactionHash!: string;
+    @PrimaryColumn({ name: 'block_hash' })
+    public blockHash!: string;
+    @PrimaryColumn({ name: 'block_number' })
+    public blockNumber!: number;
 
-    @Column({ type: 'bigint' })
+    @Column({ type: 'bigint', name: 'gas_used' })
     public gasUsed!: number;
-    @Column({ type: 'bigint' })
+    @Column({ type: 'bigint', name: 'gas_price' })
     public gasPrice!: number;
 }
diff --git a/packages/pipeline/src/scripts/merge_v2_events.ts b/packages/pipeline/src/scripts/merge_v2_events.ts
deleted file mode 100644
index 99a76aa61..000000000
--- a/packages/pipeline/src/scripts/merge_v2_events.ts
+++ /dev/null
@@ -1,81 +0,0 @@
-import { web3Factory } from '@0x/dev-utils';
-import 'reflect-metadata';
-import { Connection, createConnection } from 'typeorm';
-
-import { ExchangeEventsSource } from '../data_sources/contract-wrappers/exchange_events';
-import { ExchangeFillEvent } from '../entities';
-import { deployConfig } from '../ormconfig';
-import { parseExchangeEvents } from '../parsers/events';
-
-let connection: Connection;
-
-(async () => {
-    connection = await createConnection(deployConfig);
-    await getExchangeEventsAsync();
-    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(`Retrieved and parsed ${events.length} total events.`);
-    console.log('Saving events...');
-    for (const event of events) {
-        await eventsRepository.save(event);
-    }
-    await eventsRepository.save(events);
-    console.log('Saved events.');
-}
-
-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);
-}