aboutsummaryrefslogtreecommitdiffstats
path: root/packages/pipeline/src/scripts/merge_v2_events.ts
diff options
context:
space:
mode:
Diffstat (limited to 'packages/pipeline/src/scripts/merge_v2_events.ts')
-rw-r--r--packages/pipeline/src/scripts/merge_v2_events.ts81
1 files changed, 81 insertions, 0 deletions
diff --git a/packages/pipeline/src/scripts/merge_v2_events.ts b/packages/pipeline/src/scripts/merge_v2_events.ts
new file mode 100644
index 000000000..227ece121
--- /dev/null
+++ b/packages/pipeline/src/scripts/merge_v2_events.ts
@@ -0,0 +1,81 @@
+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/ExchangeFillEvent';
+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);
+}