Get uniswap events in block range (#139)

* Get events in block range.

* Get uniswap events in block range.

packages/address-watcher/src/entity/Account.ts

@@ -6,7 +6,7 @@ export class Account {
   @PrimaryColumn('varchar', { length: 42 })
   address!: string;
 
-  @Column('numeric')
+  @Column('integer')
   startingBlock!: number;
 
   @ManyToMany(() => Trace, trace => trace.accounts)

packages/address-watcher/src/entity/BlockProgress.ts

@@ -6,13 +6,13 @@ export class BlockProgress {
   @PrimaryColumn('varchar', { length: 66 })
   blockHash!: string;
 
-  @Column('numeric')
+  @Column('integer')
   blockNumber!: number;
 
-  @Column('numeric')
+  @Column('integer')
   numTx!: number;
 
-  @Column('numeric')
+  @Column('integer')
   numTracedTx!: number;
 
   @Column('boolean')

packages/address-watcher/src/entity/Trace.ts

@@ -9,7 +9,7 @@ export class Trace {
   @PrimaryColumn('varchar', { length: 66 })
   txHash!: string;
 
-  @Column('numeric')
+  @Column('integer')
   blockNumber!: number;
 
   @Column('varchar', { length: 66 })

packages/erc20-watcher/src/entity/Contract.ts

@@ -9,6 +9,6 @@ export class Contract {
   @Column('varchar', { length: 42 })
   address!: string;
 
-  @Column('numeric')
+  @Column('integer')
   startingBlock!: number;
 }

packages/uni-watcher/src/database.ts

@@ -1,8 +1,9 @@
 import assert from 'assert';
+import _ from 'lodash';
 import { Connection, ConnectionOptions, createConnection, DeepPartial } from 'typeorm';
 import { SnakeNamingStrategy } from 'typeorm-naming-strategies';
 
-import { Event } from './entity/Event';
+import { Event, UNKNOWN_EVENT_NAME } from './entity/Event';
 import { Contract } from './entity/Contract';
 import { BlockProgress } from './entity/BlockProgress';
 
@@ -58,6 +59,35 @@ export class Database {
       .getMany();
   }
 
+  async getProcessedBlockCountForRange (fromBlockNumber: number, toBlockNumber: number): Promise<{ expected: number, actual: number }> {
+    const blockNumbers = _.range(fromBlockNumber, toBlockNumber + 1);
+    const expected = blockNumbers.length;
+
+    const repo = this._conn.getRepository(BlockProgress);
+    const { count: actual } = await repo
+      .createQueryBuilder('block_progress')
+      .select('COUNT(DISTINCT(block_number))', 'count')
+      .where('block_number IN (:...blockNumbers) AND is_complete = :isComplete', { blockNumbers, isComplete: true })
+      .getRawOne();
+
+    return { expected, actual: parseInt(actual) };
+  }
+
+  async getEventsInRange (fromBlockNumber: number, toBlockNumber: number): Promise<Array<Event>> {
+    return this._conn.getRepository(Event)
+      .createQueryBuilder('event')
+      .where('block_number >= :fromBlockNumber AND block_number <= :toBlockNumber AND event_name <> :eventName', {
+        fromBlockNumber,
+        toBlockNumber,
+        eventName: UNKNOWN_EVENT_NAME
+      })
+      .orderBy({
+        block_number: 'ASC',
+        index: 'ASC'
+      })
+      .getMany();
+  }
+
   async saveEvents (blockHash: string, blockNumber: number, events: DeepPartial<Event>[]): Promise<void> {
     // In a transaction:
     // (1) Save all the events in the database.

packages/uni-watcher/src/entity/BlockProgress.ts

@@ -9,13 +9,13 @@ export class BlockProgress {
   @Column('varchar', { length: 66 })
   blockHash!: string;
 
-  @Column('numeric')
+  @Column('integer')
   blockNumber!: number;
 
-  @Column('numeric')
+  @Column('integer')
   numEvents!: number;
 
-  @Column('numeric')
+  @Column('integer')
   numProcessedEvents!: number;
 
   @Column('boolean')

packages/uni-watcher/src/entity/Contract.ts

@@ -15,6 +15,6 @@ export class Contract {
   @Column('varchar', { length: 8 })
   kind!: string;
 
-  @Column('numeric')
+  @Column('integer')
   startingBlock!: number;
 }

packages/uni-watcher/src/entity/Event.ts

@@ -5,6 +5,8 @@ export const UNKNOWN_EVENT_NAME = '__unknown__';
 @Entity()
 // Index to query all events for a contract efficiently.
 @Index(['blockHash', 'contract'])
+// Index to query block range for uniswap events.
+@Index(['blockNumber', 'eventName'])
 export class Event {
   @PrimaryGeneratedColumn()
   id!: number;

packages/uni-watcher/src/indexer.ts

@@ -15,6 +15,9 @@ import { Contract, KIND_FACTORY, KIND_POOL } from './entity/Contract';
 import factoryABI from './artifacts/factory.json';
 import poolABI from './artifacts/pool.json';
 
+// TODO: Move to config.
+const MAX_EVENTS_BLOCK_RANGE = 1000;
+
 const log = debug('vulcanize:indexer');
 
 type ResultEvent = {
@@ -294,4 +297,20 @@ export class Indexer {
   async updateBlockProgress (blockHash: string): Promise<void> {
     return this._db.updateBlockProgress(blockHash);
   }
+
+  async getProcessedBlockCountForRange (fromBlockNumber: number, toBlockNumber: number): Promise<{ expected: number, actual: number }> {
+    return this._db.getProcessedBlockCountForRange(fromBlockNumber, toBlockNumber);
+  }
+
+  async getEventsInRange (fromBlockNumber: number, toBlockNumber: number): Promise<Array<Event>> {
+    if (toBlockNumber <= fromBlockNumber) {
+      throw new Error('toBlockNumber should be greater than fromBlockNumber');
+    }
+
+    if ((toBlockNumber - fromBlockNumber) > MAX_EVENTS_BLOCK_RANGE) {
+      throw new Error(`Max range (${MAX_EVENTS_BLOCK_RANGE}) exceeded`);
+    }
+
+    return this._db.getEventsInRange(fromBlockNumber, toBlockNumber);
+  }
 }

packages/uni-watcher/src/resolvers.ts

@@ -66,7 +66,19 @@ export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatch
 
         const events = await indexer.getEventsByFilter(blockHash, contract, name);
         return events.map(event => indexer.getResultEvent(event));
-      }
+      },
+
+      eventsInRange: async (_: any, { fromBlockNumber, toBlockNumber }: { fromBlockNumber: number, toBlockNumber: number }) => {
+        log('eventsInRange', fromBlockNumber, toBlockNumber);
+
+        const { expected, actual } = await indexer.getProcessedBlockCountForRange(fromBlockNumber, toBlockNumber);
+        if (expected !== actual) {
+          throw new Error(`Range not available, expected ${expected}, got ${actual} blocks in range`);
+        }
+
+        const events = await indexer.getEventsInRange(fromBlockNumber, toBlockNumber);
+        return events.map(event => indexer.getResultEvent(event));
+      },
     }
   };
 };