aboutsummaryrefslogtreecommitdiffstats
path: root/packages/pipeline/src
diff options
context:
space:
mode:
Diffstat (limited to 'packages/pipeline/src')
-rw-r--r--packages/pipeline/src/data_sources/contract-wrappers/exchange_events.ts5
-rw-r--r--packages/pipeline/src/entities/ExchangeCancelEvent.ts5
-rw-r--r--packages/pipeline/src/entities/ExchangeCancelUpToEvent.ts5
-rw-r--r--packages/pipeline/src/entities/ExchangeFillEvent.ts7
-rw-r--r--packages/pipeline/src/entities/SraOrder.ts4
-rw-r--r--packages/pipeline/src/index.ts76
-rw-r--r--packages/pipeline/src/ormconfig.ts15
-rw-r--r--packages/pipeline/src/parsers/events/index.ts3
8 files changed, 84 insertions, 36 deletions
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();
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-02 20:00:55 +0800