cerc-io/watcher-ts
16
1
Fork 0
mirror of https://github.com/cerc-io/watcher-ts synced 2025-01-08 12:28:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

Event handlers should update db atomically (#186)

Browse Source 
* Run db operations inside event handlers atomically using transaction.

* Implement database transaction for Pool Initialize event.

* Implement typeorm transaction without callback.

* Implement transaction for NFPM event handlers.

Co-authored-by: nabarun <nabarun@deepstacksoft.com>
This commit is contained in:
Ashwin Phatak 2021-08-06 10:25:56 +05:30 committed by GitHub
parent b6fe8a8c47
commit 0487c05ee1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 889 additions and 727 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

267
packages/uni-info-watcher/src/database.ts
View File

@ -1,5 +1,5 @@
import assert from 'assert'; import assert from 'assert';
import { Brackets, Connection, ConnectionOptions, createConnection, DeepPartial, FindConditions, FindOneOptions, In, LessThanOrEqual, Repository } from 'typeorm'; import { Brackets, Connection, ConnectionOptions, createConnection, DeepPartial, FindConditions, FindOneOptions, In, LessThanOrEqual, QueryRunner, Repository } from 'typeorm';
import { SnakeNamingStrategy } from 'typeorm-naming-strategies'; import { SnakeNamingStrategy } from 'typeorm-naming-strategies';
import { MAX_REORG_DEPTH } from '@vulcanize/util'; import { MAX_REORG_DEPTH } from '@vulcanize/util';
@ -87,8 +87,15 @@ export class Database {
return this._conn.close(); return this._conn.close();
} }
async getFactory ({ id, blockHash }: DeepPartial<Factory>): Promise<Factory | undefined> { async createTransactionRunner (): Promise<QueryRunner> {
const repo = this._conn.getRepository(Factory); const queryRunner = this._conn.createQueryRunner();
await queryRunner.connect();
await queryRunner.startTransaction();
return queryRunner;
}
async getFactory (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<Factory>): Promise<Factory | undefined> {
const repo = queryRunner.manager.getRepository(Factory);
const whereOptions: FindConditions<Factory> = { id }; const whereOptions: FindConditions<Factory> = { id };
if (blockHash) { if (blockHash) {
@ -111,8 +118,8 @@ export class Database {
return entity; return entity;
} }
async getBundle ({ id, blockHash, blockNumber }: DeepPartial<Bundle>): Promise<Bundle | undefined> { async getBundle (queryRunner: QueryRunner, { id, blockHash, blockNumber }: DeepPartial<Bundle>): Promise<Bundle | undefined> {
const repo = this._conn.getRepository(Bundle); const repo = queryRunner.manager.getRepository(Bundle);
const whereOptions: FindConditions<Bundle> = { id }; const whereOptions: FindConditions<Bundle> = { id };
if (blockHash) { if (blockHash) {
@ -139,8 +146,8 @@ export class Database {
return entity; return entity;
} }
async getToken ({ id, blockHash }: DeepPartial<Token>): Promise<Token | undefined> { async getToken (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<Token>): Promise<Token | undefined> {
const repo = this._conn.getRepository(Token); const repo = queryRunner.manager.getRepository(Token);
const whereOptions: FindConditions<Token> = { id }; const whereOptions: FindConditions<Token> = { id };
if (blockHash) { if (blockHash) {
@ -164,8 +171,14 @@ export class Database {
return entity; return entity;
} }
async getPool ({ id, blockHash, blockNumber }: DeepPartial<Pool>): Promise<Pool | undefined> { async getTokenNoTx ({ id, blockHash }: DeepPartial<Token>): Promise<Token | undefined> {
const repo = this._conn.getRepository(Pool); const queryRunner = this._conn.createQueryRunner();
await queryRunner.connect();
return this.getToken(queryRunner, { id, blockHash });
}
async getPool (queryRunner: QueryRunner, { id, blockHash, blockNumber }: DeepPartial<Pool>): Promise<Pool | undefined> {
const repo = queryRunner.manager.getRepository(Pool);
const whereOptions: FindConditions<Pool> = { id }; const whereOptions: FindConditions<Pool> = { id };
if (blockHash) { if (blockHash) {
@ -193,6 +206,12 @@ export class Database {
return entity; return entity;
} }
async getPoolNoTx ({ id, blockHash, blockNumber }: DeepPartial<Pool>): Promise<Pool | undefined> {
const queryRunner = this._conn.createQueryRunner();
await queryRunner.connect();
return this.getPool(queryRunner, { id, blockHash, blockNumber });
}
async getPosition ({ id, blockHash }: DeepPartial<Position>): Promise<Position | undefined> { async getPosition ({ id, blockHash }: DeepPartial<Position>): Promise<Position | undefined> {
const repo = this._conn.getRepository(Position); const repo = this._conn.getRepository(Position);
const whereOptions: FindConditions<Position> = { id }; const whereOptions: FindConditions<Position> = { id };
@ -218,8 +237,8 @@ export class Database {
return entity; return entity;
} }
async getTick ({ id, blockHash }: DeepPartial<Tick>): Promise<Tick | undefined> { async getTick (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<Tick>): Promise<Tick | undefined> {
const repo = this._conn.getRepository(Tick); const repo = queryRunner.manager.getRepository(Tick);
const whereOptions: FindConditions<Tick> = { id }; const whereOptions: FindConditions<Tick> = { id };
if (blockHash) { if (blockHash) {
@ -243,8 +262,14 @@ export class Database {
return entity; return entity;
} }
async getPoolDayData ({ id, blockHash }: DeepPartial<PoolDayData>): Promise<PoolDayData | undefined> { async getTickNoTx ({ id, blockHash }: DeepPartial<Tick>): Promise<Tick | undefined> {
const repo = this._conn.getRepository(PoolDayData); const queryRunner = this._conn.createQueryRunner();
await queryRunner.connect();
return this.getTick(queryRunner, { id, blockHash });
}
async getPoolDayData (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<PoolDayData>): Promise<PoolDayData | undefined> {
const repo = queryRunner.manager.getRepository(PoolDayData);
const whereOptions: FindConditions<PoolDayData> = { id }; const whereOptions: FindConditions<PoolDayData> = { id };
if (blockHash) { if (blockHash) {
@ -268,8 +293,8 @@ export class Database {
return entity; return entity;
} }
async getPoolHourData ({ id, blockHash }: DeepPartial<PoolHourData>): Promise<PoolHourData | undefined> { async getPoolHourData (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<PoolHourData>): Promise<PoolHourData | undefined> {
const repo = this._conn.getRepository(PoolHourData); const repo = queryRunner.manager.getRepository(PoolHourData);
const whereOptions: FindConditions<PoolHourData> = { id }; const whereOptions: FindConditions<PoolHourData> = { id };
if (blockHash) { if (blockHash) {
@ -292,8 +317,8 @@ export class Database {
return entity; return entity;
} }
async getUniswapDayData ({ id, blockHash }: DeepPartial<UniswapDayData>): Promise<UniswapDayData | undefined> { async getUniswapDayData (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<UniswapDayData>): Promise<UniswapDayData | undefined> {
const repo = this._conn.getRepository(UniswapDayData); const repo = queryRunner.manager.getRepository(UniswapDayData);
const whereOptions: FindConditions<UniswapDayData> = { id }; const whereOptions: FindConditions<UniswapDayData> = { id };
if (blockHash) { if (blockHash) {
@ -316,8 +341,8 @@ export class Database {
return entity; return entity;
} }
async getTokenDayData ({ id, blockHash }: DeepPartial<TokenDayData>): Promise<TokenDayData | undefined> { async getTokenDayData (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<TokenDayData>): Promise<TokenDayData | undefined> {
const repo = this._conn.getRepository(TokenDayData); const repo = queryRunner.manager.getRepository(TokenDayData);
const whereOptions: FindConditions<TokenDayData> = { id }; const whereOptions: FindConditions<TokenDayData> = { id };
if (blockHash) { if (blockHash) {
@ -340,8 +365,8 @@ export class Database {
return entity; return entity;
} }
async getTokenHourData ({ id, blockHash }: DeepPartial<TokenHourData>): Promise<TokenHourData | undefined> { async getTokenHourData (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<TokenHourData>): Promise<TokenHourData | undefined> {
const repo = this._conn.getRepository(TokenHourData); const repo = queryRunner.manager.getRepository(TokenHourData);
const whereOptions: FindConditions<TokenHourData> = { id }; const whereOptions: FindConditions<TokenHourData> = { id };
if (blockHash) { if (blockHash) {
@ -364,8 +389,8 @@ export class Database {
return entity; return entity;
} }
async getTransaction ({ id, blockHash }: DeepPartial<Transaction>): Promise<Transaction | undefined> { async getTransaction (queryRunner: QueryRunner, { id, blockHash }: DeepPartial<Transaction>): Promise<Transaction | undefined> {
const repo = this._conn.getRepository(Transaction); const repo = queryRunner.manager.getRepository(Transaction);
const whereOptions: FindConditions<Transaction> = { id }; const whereOptions: FindConditions<Transaction> = { id };
if (blockHash) { if (blockHash) {
@ -388,8 +413,8 @@ export class Database {
return entity; return entity;
} }
async getEntities<Entity> (entity: new () => Entity, block: BlockHeight, where: Where = {}, queryOptions: QueryOptions = {}, relations: string[] = []): Promise<Entity[]> { async getEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, block: BlockHeight, where: Where = {}, queryOptions: QueryOptions = {}, relations: string[] = []): Promise<Entity[]> {
const repo = this._conn.getRepository(entity); const repo = queryRunner.manager.getRepository(entity);
const { tableName } = repo.metadata; const { tableName } = repo.metadata;
let subQuery = repo.createQueryBuilder('subTable') let subQuery = repo.createQueryBuilder('subTable')
@ -468,148 +493,116 @@ export class Database {
return selectQueryBuilder.getMany(); return selectQueryBuilder.getMany();
} }
async saveFactory (factory: Factory, block: Block): Promise<Factory> { async saveFactory (queryRunner: QueryRunner, factory: Factory, block: Block): Promise<Factory> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Factory);
const repo = tx.getRepository(Factory); factory.blockNumber = block.number;
factory.blockNumber = block.number; factory.blockHash = block.hash;
factory.blockHash = block.hash; return repo.save(factory);
return repo.save(factory);
});
} }
async saveBundle (bundle: Bundle, block: Block): Promise<Bundle> { async saveBundle (queryRunner: QueryRunner, bundle: Bundle, block: Block): Promise<Bundle> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Bundle);
const repo = tx.getRepository(Bundle); bundle.blockNumber = block.number;
bundle.blockNumber = block.number; bundle.blockHash = block.hash;
bundle.blockHash = block.hash; return repo.save(bundle);
return repo.save(bundle);
});
} }
async savePool (pool: Pool, block: Block): Promise<Pool> { async savePool (queryRunner: QueryRunner, pool: Pool, block: Block): Promise<Pool> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Pool);
const repo = tx.getRepository(Pool); pool.blockNumber = block.number;
pool.blockNumber = block.number; pool.blockHash = block.hash;
pool.blockHash = block.hash; return repo.save(pool);
return repo.save(pool);
});
} }
async savePoolDayData (poolDayData: PoolDayData, block: Block): Promise<PoolDayData> { async savePoolDayData (queryRunner: QueryRunner, poolDayData: PoolDayData, block: Block): Promise<PoolDayData> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(PoolDayData);
const repo = tx.getRepository(PoolDayData); poolDayData.blockNumber = block.number;
poolDayData.blockNumber = block.number; poolDayData.blockHash = block.hash;
poolDayData.blockHash = block.hash; return repo.save(poolDayData);
return repo.save(poolDayData);
});
} }
async savePoolHourData (poolHourData: PoolHourData, block: Block): Promise<PoolHourData> { async savePoolHourData (queryRunner: QueryRunner, poolHourData: PoolHourData, block: Block): Promise<PoolHourData> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(PoolHourData);
const repo = tx.getRepository(PoolHourData); poolHourData.blockNumber = block.number;
poolHourData.blockNumber = block.number; poolHourData.blockHash = block.hash;
poolHourData.blockHash = block.hash; return repo.save(poolHourData);
return repo.save(poolHourData);
});
} }
async saveToken (token: Token, block: Block): Promise<Token> { async saveToken (queryRunner: QueryRunner, token: Token, block: Block): Promise<Token> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Token);
const repo = tx.getRepository(Token); token.blockNumber = block.number;
token.blockNumber = block.number; token.blockHash = block.hash;
token.blockHash = block.hash; return repo.save(token);
return repo.save(token);
});
} }
async saveTransaction (transaction: Transaction, block: Block): Promise<Transaction> { async saveTransaction (queryRunner: QueryRunner, transaction: Transaction, block: Block): Promise<Transaction> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Transaction);
const repo = tx.getRepository(Transaction); transaction.blockNumber = block.number;
transaction.blockNumber = block.number; transaction.blockHash = block.hash;
transaction.blockHash = block.hash; return repo.save(transaction);
return repo.save(transaction);
});
} }
async saveUniswapDayData (uniswapDayData: UniswapDayData, block: Block): Promise<UniswapDayData> { async saveUniswapDayData (queryRunner: QueryRunner, uniswapDayData: UniswapDayData, block: Block): Promise<UniswapDayData> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(UniswapDayData);
const repo = tx.getRepository(UniswapDayData); uniswapDayData.blockNumber = block.number;
uniswapDayData.blockNumber = block.number; uniswapDayData.blockHash = block.hash;
uniswapDayData.blockHash = block.hash; return repo.save(uniswapDayData);
return repo.save(uniswapDayData);
});
} }
async saveTokenDayData (tokenDayData: TokenDayData, block: Block): Promise<TokenDayData> { async saveTokenDayData (queryRunner: QueryRunner, tokenDayData: TokenDayData, block: Block): Promise<TokenDayData> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(TokenDayData);
const repo = tx.getRepository(TokenDayData); tokenDayData.blockNumber = block.number;
tokenDayData.blockNumber = block.number; tokenDayData.blockHash = block.hash;
tokenDayData.blockHash = block.hash; return repo.save(tokenDayData);
return repo.save(tokenDayData);
});
} }
async saveTokenHourData (tokenHourData: TokenHourData, block: Block): Promise<TokenHourData> { async saveTokenHourData (queryRunner: QueryRunner, tokenHourData: TokenHourData, block: Block): Promise<TokenHourData> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(TokenHourData);
const repo = tx.getRepository(TokenHourData); tokenHourData.blockNumber = block.number;
tokenHourData.blockNumber = block.number; tokenHourData.blockHash = block.hash;
tokenHourData.blockHash = block.hash; return repo.save(tokenHourData);
return repo.save(tokenHourData);
});
} }
async saveTick (tick: Tick, block: Block): Promise<Tick> { async saveTick (queryRunner: QueryRunner, tick: Tick, block: Block): Promise<Tick> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Tick);
const repo = tx.getRepository(Tick); tick.blockNumber = block.number;
tick.blockNumber = block.number; tick.blockHash = block.hash;
tick.blockHash = block.hash; return repo.save(tick);
return repo.save(tick);
});
} }
async savePosition (position: Position, block: Block): Promise<Position> { async savePosition (queryRunner: QueryRunner, position: Position, block: Block): Promise<Position> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Position);
const repo = tx.getRepository(Position); position.blockNumber = block.number;
position.blockNumber = block.number; position.blockHash = block.hash;
position.blockHash = block.hash; return repo.save(position);
return repo.save(position);
});
} }
async savePositionSnapshot (positionSnapshot: PositionSnapshot, block: Block): Promise<PositionSnapshot> { async savePositionSnapshot (queryRunner: QueryRunner, positionSnapshot: PositionSnapshot, block: Block): Promise<PositionSnapshot> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(PositionSnapshot);
const repo = tx.getRepository(PositionSnapshot); positionSnapshot.blockNumber = block.number;
positionSnapshot.blockNumber = block.number; positionSnapshot.blockHash = block.hash;
positionSnapshot.blockHash = block.hash; return repo.save(positionSnapshot);
return repo.save(positionSnapshot);
});
} }
async saveMint (mint: Mint, block: Block): Promise<Mint> { async saveMint (queryRunner: QueryRunner, mint: Mint, block: Block): Promise<Mint> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Mint);
const repo = tx.getRepository(Mint); mint.blockNumber = block.number;
mint.blockNumber = block.number; mint.blockHash = block.hash;
mint.blockHash = block.hash; return repo.save(mint);
return repo.save(mint);
});
} }
async saveBurn (burn: Burn, block: Block): Promise<Burn> { async saveBurn (queryRunner: QueryRunner, burn: Burn, block: Block): Promise<Burn> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Burn);
const repo = tx.getRepository(Burn); burn.blockNumber = block.number;
burn.blockNumber = block.number; burn.blockHash = block.hash;
burn.blockHash = block.hash; return repo.save(burn);
return repo.save(burn);
});
} }
async saveSwap (swap: Swap, block: Block): Promise<Swap> { async saveSwap (queryRunner: QueryRunner, swap: Swap, block: Block): Promise<Swap> {
return this._conn.transaction(async (tx) => { const repo = queryRunner.manager.getRepository(Swap);
const repo = tx.getRepository(Swap); swap.blockNumber = block.number;
swap.blockNumber = block.number; swap.blockHash = block.hash;
swap.blockHash = block.hash; return repo.save(swap);
return repo.save(swap);
});
} }
// Returns true if events have already been synced for the (block, token) combination. // Returns true if events have already been synced for the (block, token) combination.

1284
packages/uni-info-watcher/src/indexer.ts
View File

File diff suppressed because it is too large Load Diff

7
packages/uni-info-watcher/src/utils/index.ts
View File

@ -1,5 +1,6 @@
import Decimal from 'decimal.js'; import Decimal from 'decimal.js';
import { BigNumber } from 'ethers'; import { BigNumber } from 'ethers';
import { QueryRunner } from 'typeorm';
import { Transaction as TransactionEntity } from '../entity/Transaction'; import { Transaction as TransactionEntity } from '../entity/Transaction';
import { Database } from '../database'; import { Database } from '../database';
@ -23,9 +24,9 @@ export const convertTokenToDecimal = (tokenAmount: bigint, exchangeDecimals: big
return (new Decimal(tokenAmount.toString())).div(exponentToBigDecimal(exchangeDecimals)); return (new Decimal(tokenAmount.toString())).div(exponentToBigDecimal(exchangeDecimals));
}; };
export const loadTransaction = async (db: Database, event: { block: Block, tx: Transaction }): Promise<TransactionEntity> => { export const loadTransaction = async (db: Database, dbTx: QueryRunner, event: { block: Block, tx: Transaction }): Promise<TransactionEntity> => {
const { tx, block } = event; const { tx, block } = event;
let transaction = await db.getTransaction({ id: tx.hash, blockHash: block.hash }); let transaction = await db.getTransaction(dbTx, { id: tx.hash, blockHash: block.hash });
if (!transaction) { if (!transaction) {
transaction = new TransactionEntity(); transaction = new TransactionEntity();
@ -35,7 +36,7 @@ export const loadTransaction = async (db: Database, event: { block: Block, tx: T
transaction.blockNumber = block.number; transaction.blockNumber = block.number;
transaction.timestamp = BigInt(block.timestamp); transaction.timestamp = BigInt(block.timestamp);
return db.saveTransaction(transaction, block); return db.saveTransaction(dbTx, transaction, block);
}; };
// Return 0 if denominator is 0 in division. // Return 0 if denominator is 0 in division.

45
packages/uni-info-watcher/src/utils/interval-updates.ts
View File

@ -1,5 +1,6 @@
import assert from 'assert'; import assert from 'assert';
import { BigNumber } from 'ethers'; import { BigNumber } from 'ethers';
import { QueryRunner } from 'typeorm';
import { Database } from '../database'; import { Database } from '../database';
import { Factory } from '../entity/Factory'; import { Factory } from '../entity/Factory';
@ -16,17 +17,17 @@ import { Block } from '../events';
* @param db * @param db
* @param event * @param event
*/ */
export const updateUniswapDayData = async (db: Database, event: { contractAddress: string, block: Block }): Promise<UniswapDayData> => { export const updateUniswapDayData = async (db: Database, dbTx: QueryRunner, event: { contractAddress: string, block: Block }): Promise<UniswapDayData> => {
const { block } = event; const { block } = event;
// TODO: In subgraph factory is fetched by hardcoded factory address. // TODO: In subgraph factory is fetched by hardcoded factory address.
// Currently fetching first factory in database as only one exists. // Currently fetching first factory in database as only one exists.
const [factory] = await db.getEntities(Factory, { hash: block.hash }, {}, { limit: 1 }); const [factory] = await db.getEntities(dbTx, Factory, { hash: block.hash }, {}, { limit: 1 });
const dayID = Math.floor(block.timestamp / 86400); // Rounded. const dayID = Math.floor(block.timestamp / 86400); // Rounded.
const dayStartTimestamp = dayID * 86400; const dayStartTimestamp = dayID * 86400;
let uniswapDayData = await db.getUniswapDayData({ id: dayID.toString(), blockHash: block.hash }); let uniswapDayData = await db.getUniswapDayData(dbTx, { id: dayID.toString(), blockHash: block.hash });
if (!uniswapDayData) { if (!uniswapDayData) {
uniswapDayData = new UniswapDayData(); uniswapDayData = new UniswapDayData();
@ -38,10 +39,10 @@ export const updateUniswapDayData = async (db: Database, event: { contractAddres
uniswapDayData.tvlUSD = factory.totalValueLockedUSD; uniswapDayData.tvlUSD = factory.totalValueLockedUSD;
uniswapDayData.txCount = factory.txCount; uniswapDayData.txCount = factory.txCount;
return db.saveUniswapDayData(uniswapDayData, block); return db.saveUniswapDayData(dbTx, uniswapDayData, block);
}; };
export const updatePoolDayData = async (db: Database, event: { contractAddress: string, block: Block }): Promise<PoolDayData> => { export const updatePoolDayData = async (db: Database, dbTx: QueryRunner, event: { contractAddress: string, block: Block }): Promise<PoolDayData> => {
const { contractAddress, block } = event; const { contractAddress, block } = event;
const dayID = Math.floor(block.timestamp / 86400); const dayID = Math.floor(block.timestamp / 86400);
const dayStartTimestamp = dayID * 86400; const dayStartTimestamp = dayID * 86400;
@ -50,10 +51,10 @@ export const updatePoolDayData = async (db: Database, event: { contractAddress:
.concat('-') .concat('-')
.concat(dayID.toString()); .concat(dayID.toString());
const pool = await db.getPool({ id: contractAddress, blockHash: block.hash }); const pool = await db.getPool(dbTx, { id: contractAddress, blockHash: block.hash });
assert(pool); assert(pool);
let poolDayData = await db.getPoolDayData({ id: dayPoolID, blockHash: block.hash }); let poolDayData = await db.getPoolDayData(dbTx, { id: dayPoolID, blockHash: block.hash });
if (!poolDayData) { if (!poolDayData) {
poolDayData = new PoolDayData(); poolDayData = new PoolDayData();
@ -64,7 +65,7 @@ export const updatePoolDayData = async (db: Database, event: { contractAddress:
poolDayData.high = pool.token0Price; poolDayData.high = pool.token0Price;
poolDayData.low = pool.token0Price; poolDayData.low = pool.token0Price;
poolDayData.close = pool.token0Price; poolDayData.close = pool.token0Price;
poolDayData = await db.savePoolDayData(poolDayData, block); poolDayData = await db.savePoolDayData(dbTx, poolDayData, block);
} }
if (Number(pool.token0Price) > Number(poolDayData.high)) { if (Number(pool.token0Price) > Number(poolDayData.high)) {
@ -84,12 +85,12 @@ export const updatePoolDayData = async (db: Database, event: { contractAddress:
poolDayData.tick = pool.tick; poolDayData.tick = pool.tick;
poolDayData.tvlUSD = pool.totalValueLockedUSD; poolDayData.tvlUSD = pool.totalValueLockedUSD;
poolDayData.txCount = BigInt(BigNumber.from(poolDayData.txCount).add(1).toHexString()); poolDayData.txCount = BigInt(BigNumber.from(poolDayData.txCount).add(1).toHexString());
poolDayData = await db.savePoolDayData(poolDayData, block); poolDayData = await db.savePoolDayData(dbTx, poolDayData, block);
return poolDayData; return poolDayData;
}; };
export const updatePoolHourData = async (db: Database, event: { contractAddress: string, block: Block }): Promise<PoolHourData> => { export const updatePoolHourData = async (db: Database, dbTx: QueryRunner, event: { contractAddress: string, block: Block }): Promise<PoolHourData> => {
const { contractAddress, block } = event; const { contractAddress, block } = event;
const hourIndex = Math.floor(block.timestamp / 3600); // Get unique hour within unix history. const hourIndex = Math.floor(block.timestamp / 3600); // Get unique hour within unix history.
const hourStartUnix = hourIndex * 3600; // Want the rounded effect. const hourStartUnix = hourIndex * 3600; // Want the rounded effect.
@ -98,10 +99,10 @@ export const updatePoolHourData = async (db: Database, event: { contractAddress:
.concat('-') .concat('-')
.concat(hourIndex.toString()); .concat(hourIndex.toString());
const pool = await db.getPool({ id: contractAddress, blockHash: block.hash }); const pool = await db.getPool(dbTx, { id: contractAddress, blockHash: block.hash });
assert(pool); assert(pool);
let poolHourData = await db.getPoolHourData({ id: hourPoolID, blockHash: block.hash }); let poolHourData = await db.getPoolHourData(dbTx, { id: hourPoolID, blockHash: block.hash });
if (!poolHourData) { if (!poolHourData) {
poolHourData = new PoolHourData(); poolHourData = new PoolHourData();
@ -112,7 +113,7 @@ export const updatePoolHourData = async (db: Database, event: { contractAddress:
poolHourData.high = pool.token0Price; poolHourData.high = pool.token0Price;
poolHourData.low = pool.token0Price; poolHourData.low = pool.token0Price;
poolHourData.close = pool.token0Price; poolHourData.close = pool.token0Price;
poolHourData = await db.savePoolHourData(poolHourData, block); poolHourData = await db.savePoolHourData(dbTx, poolHourData, block);
} }
if (Number(pool.token0Price) > Number(poolHourData.high)) { if (Number(pool.token0Price) > Number(poolHourData.high)) {
@ -132,14 +133,14 @@ export const updatePoolHourData = async (db: Database, event: { contractAddress:
poolHourData.tick = pool.tick; poolHourData.tick = pool.tick;
poolHourData.tvlUSD = pool.totalValueLockedUSD; poolHourData.tvlUSD = pool.totalValueLockedUSD;
poolHourData.txCount = BigInt(BigNumber.from(poolHourData.txCount).add(1).toHexString()); poolHourData.txCount = BigInt(BigNumber.from(poolHourData.txCount).add(1).toHexString());
poolHourData = await db.savePoolHourData(poolHourData, block); poolHourData = await db.savePoolHourData(dbTx, poolHourData, block);
return poolHourData; return poolHourData;
}; };
export const updateTokenDayData = async (db: Database, token: Token, event: { block: Block }): Promise<TokenDayData> => { export const updateTokenDayData = async (db: Database, dbTx: QueryRunner, token: Token, event: { block: Block }): Promise<TokenDayData> => {
const { block } = event; const { block } = event;
const bundle = await db.getBundle({ id: '1', blockHash: block.hash }); const bundle = await db.getBundle(dbTx, { id: '1', blockHash: block.hash });
assert(bundle); assert(bundle);
const dayID = Math.floor(block.timestamp / 86400); const dayID = Math.floor(block.timestamp / 86400);
const dayStartTimestamp = dayID * 86400; const dayStartTimestamp = dayID * 86400;
@ -150,7 +151,7 @@ export const updateTokenDayData = async (db: Database, token: Token, event: { bl
const tokenPrice = token.derivedETH.times(bundle.ethPriceUSD); const tokenPrice = token.derivedETH.times(bundle.ethPriceUSD);
let tokenDayData = await db.getTokenDayData({ id: tokenDayID, blockHash: block.hash }); let tokenDayData = await db.getTokenDayData(dbTx, { id: tokenDayID, blockHash: block.hash });
if (!tokenDayData) { if (!tokenDayData) {
tokenDayData = new TokenDayData(); tokenDayData = new TokenDayData();
@ -178,12 +179,12 @@ export const updateTokenDayData = async (db: Database, token: Token, event: { bl
tokenDayData.priceUSD = token.derivedETH.times(bundle.ethPriceUSD); tokenDayData.priceUSD = token.derivedETH.times(bundle.ethPriceUSD);
tokenDayData.totalValueLocked = token.totalValueLocked; tokenDayData.totalValueLocked = token.totalValueLocked;
tokenDayData.totalValueLockedUSD = token.totalValueLockedUSD; tokenDayData.totalValueLockedUSD = token.totalValueLockedUSD;
return db.saveTokenDayData(tokenDayData, block); return db.saveTokenDayData(dbTx, tokenDayData, block);
}; };
export const updateTokenHourData = async (db: Database, token: Token, event: { block: Block }): Promise<TokenHourData> => { export const updateTokenHourData = async (db: Database, dbTx: QueryRunner, token: Token, event: { block: Block }): Promise<TokenHourData> => {
const { block } = event; const { block } = event;
const bundle = await db.getBundle({ id: '1', blockHash: block.hash }); const bundle = await db.getBundle(dbTx, { id: '1', blockHash: block.hash });
assert(bundle); assert(bundle);
const hourIndex = Math.floor(block.timestamp / 3600); // Get unique hour within unix history. const hourIndex = Math.floor(block.timestamp / 3600); // Get unique hour within unix history.
const hourStartUnix = hourIndex * 3600; // Want the rounded effect. const hourStartUnix = hourIndex * 3600; // Want the rounded effect.
@ -194,7 +195,7 @@ export const updateTokenHourData = async (db: Database, token: Token, event: { b
const tokenPrice = token.derivedETH.times(bundle.ethPriceUSD); const tokenPrice = token.derivedETH.times(bundle.ethPriceUSD);
let tokenHourData = await db.getTokenHourData({ id: tokenHourID, blockHash: block.hash }); let tokenHourData = await db.getTokenHourData(dbTx, { id: tokenHourID, blockHash: block.hash });
if (!tokenHourData) { if (!tokenHourData) {
tokenHourData = new TokenHourData(); tokenHourData = new TokenHourData();
@ -222,5 +223,5 @@ export const updateTokenHourData = async (db: Database, token: Token, event: { b
tokenHourData.priceUSD = tokenPrice; tokenHourData.priceUSD = tokenPrice;
tokenHourData.totalValueLocked = token.totalValueLocked; tokenHourData.totalValueLocked = token.totalValueLocked;
tokenHourData.totalValueLockedUSD = token.totalValueLockedUSD; tokenHourData.totalValueLockedUSD = token.totalValueLockedUSD;
return db.saveTokenHourData(tokenHourData, block); return db.saveTokenHourData(dbTx, tokenHourData, block);
}; };

8
packages/uni-info-watcher/src/utils/pricing.ts
View File

@ -1,6 +1,7 @@
import assert from 'assert'; import assert from 'assert';
import Decimal from 'decimal.js'; import Decimal from 'decimal.js';
import { BigNumber } from 'ethers'; import { BigNumber } from 'ethers';
import { QueryRunner } from 'typeorm';
import { exponentToBigDecimal, safeDiv } from '.'; import { exponentToBigDecimal, safeDiv } from '.';
import { Database } from '../database'; import { Database } from '../database';
@ -55,9 +56,9 @@ export const sqrtPriceX96ToTokenPrices = (sqrtPriceX96: bigint, token0: Token, t
return [price0, price1]; return [price0, price1];
}; };
export const getEthPriceInUSD = async (db: Database, block: Block): Promise<Decimal> => { export const getEthPriceInUSD = async (db: Database, dbTx: QueryRunner, block: Block): Promise<Decimal> => {
// Fetch eth prices for each stablecoin. // Fetch eth prices for each stablecoin.
const usdcPool = await db.getPool({ id: USDC_WETH_03_POOL, blockHash: block.hash }); // DAI is token0. const usdcPool = await db.getPool(dbTx, { id: USDC_WETH_03_POOL, blockHash: block.hash }); // DAI is token0.
if (usdcPool) { if (usdcPool) {
return usdcPool.token0Price; return usdcPool.token0Price;
@ -122,12 +123,13 @@ export const findEthPerToken = async (token: Token): Promise<Decimal> => {
*/ */
export const getTrackedAmountUSD = async ( export const getTrackedAmountUSD = async (
db: Database, db: Database,
dbTx: QueryRunner,
tokenAmount0: Decimal, tokenAmount0: Decimal,
token0: Token, token0: Token,
tokenAmount1: Decimal, tokenAmount1: Decimal,
token1: Token token1: Token
): Promise<Decimal> => { ): Promise<Decimal> => {
const bundle = await db.getBundle({ id: '1' }); const bundle = await db.getBundle(dbTx, { id: '1' });
assert(bundle); assert(bundle);
const price0USD = token0.derivedETH.times(bundle.ethPriceUSD); const price0USD = token0.derivedETH.times(bundle.ethPriceUSD);
const price1USD = token1.derivedETH.times(bundle.ethPriceUSD); const price1USD = token1.derivedETH.times(bundle.ethPriceUSD);

5
packages/uni-info-watcher/src/utils/tick.ts
View File

@ -1,4 +1,5 @@
import Decimal from 'decimal.js'; import Decimal from 'decimal.js';
import { QueryRunner } from 'typeorm';
import { Pool } from '../entity/Pool'; import { Pool } from '../entity/Pool';
import { Database } from '../database'; import { Database } from '../database';
@ -6,7 +7,7 @@ import { bigDecimalExponated, safeDiv } from '.';
import { Tick } from '../entity/Tick'; import { Tick } from '../entity/Tick';
import { Block } from '../events'; import { Block } from '../events';
export const createTick = async (db: Database, tickId: string, tickIdx: bigint, pool: Pool, block: Block): Promise<Tick> => { export const createTick = async (db: Database, dbTx: QueryRunner, tickId: string, tickIdx: bigint, pool: Pool, block: Block): Promise<Tick> => {
const tick = new Tick(); const tick = new Tick();
tick.id = tickId; tick.id = tickId;
tick.tickIdx = tickIdx; tick.tickIdx = tickIdx;
@ -19,5 +20,5 @@ export const createTick = async (db: Database, tickId: string, tickIdx: bigint,
tick.price0 = price0; tick.price0 = price0;
tick.price1 = safeDiv(new Decimal(1), price0); tick.price1 = safeDiv(new Decimal(1), price0);
return db.saveTick(tick, block); return db.saveTick(dbTx, tick, block);
}; };