-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor price lookups to use workers
Signed-off-by: david <[email protected]>
- Loading branch information
Showing
11 changed files
with
277 additions
and
107 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,172 @@ | ||
| import Redis from "ioredis"; | ||
| import winston from "winston"; | ||
| import { Job, Worker } from "bullmq"; | ||
| import { DataSource, entities } from "@repo/indexer-database"; | ||
| import { IndexerQueues } from "./service"; | ||
| import { | ||
| getIntegratorId, | ||
| yesterday, | ||
| CoingeckoClient, | ||
| findTokenByAddress, | ||
| } from "../utils"; | ||
| import { RetryProvidersFactory } from "../web3/RetryProvidersFactory"; | ||
|
|
||
| export type PriceMessage = { | ||
| depositId: number; | ||
| originChainId: number; | ||
| }; | ||
|
|
||
| /** | ||
| * This worker listens to the `IntegratorId` queue and processes each job by: | ||
| * - Retrieving the deposit information from the database based on the provided relay hash. | ||
| * - Checking if the deposit record already has an integrator ID. | ||
| * - If the integrator ID is not set, the worker fetches it from the transaction data. | ||
| * - If found, the integrator ID is saved back into the deposit record in the database. | ||
| */ | ||
| export class PriceWorker { | ||
| public worker: Worker; | ||
| private coingeckoClient: CoingeckoClient; | ||
|
|
||
| constructor( | ||
| private redis: Redis, | ||
| private postgres: DataSource, | ||
| private logger: winston.Logger, | ||
| ) { | ||
| this.coingeckoClient = new CoingeckoClient(); | ||
| this.setWorker(); | ||
| } | ||
|
|
||
| public setWorker() { | ||
| this.worker = new Worker( | ||
| IndexerQueues.PriceQuery, | ||
| async (job: Job<PriceMessage>) => { | ||
| try { | ||
| await this.run(job.data); | ||
| } catch (error) { | ||
| this.logger.error({ | ||
| at: "PriceWorker", | ||
| message: `Error getting price for deposit ${job.data.depositId} on chain ${job.data.originChainId}`, | ||
| error, | ||
| }); | ||
| throw error; | ||
| } | ||
| }, | ||
| { connection: this.redis, concurrency: 10 }, | ||
| ); | ||
| } | ||
| private async run(params: PriceMessage) { | ||
| const { depositId, originChainId } = params; | ||
| const relayHashInfoRepository = this.postgres.getRepository( | ||
| entities.RelayHashInfo, | ||
| ); | ||
| const depositRepository = this.postgres.getRepository( | ||
| entities.V3FundsDeposited, | ||
| ); | ||
| const historicPriceRepository = this.postgres.getRepository( | ||
| entities.HistoricPrice, | ||
| ); | ||
|
|
||
| const relayHashInfo = await relayHashInfoRepository.findOne({ | ||
| where: { depositId, originChainId }, | ||
| }); | ||
| const deposit = await depositRepository.findOne({ | ||
| where: { depositId, originChainId }, | ||
| }); | ||
|
|
||
| if (!relayHashInfo || !deposit) { | ||
| this.logger.error({ | ||
| at: "PriceWorker", | ||
| message: "Relay hash info not found", | ||
| ...params, | ||
| }); | ||
| return; | ||
| } | ||
|
|
||
| const blockTime = relayHashInfo?.depositEvent?.blockTimestamp; | ||
| if (!blockTime) { | ||
| this.logger.error({ | ||
| at: "PriceWorker", | ||
| message: "Deposit block time not found for relay hash info", | ||
| ...params, | ||
| }); | ||
| return; | ||
| } | ||
| const priceTime = yesterday(blockTime); | ||
| const quoteCurrency = "usd"; | ||
| const baseTokenInfo = findTokenByAddress( | ||
| relayHashInfo.fillEvent.outputToken, | ||
| relayHashInfo.destinationChainId, | ||
| ); | ||
| const baseCurrency = baseTokenInfo?.coingeckoId; | ||
| let price: undefined | number; | ||
|
|
||
| if (!baseCurrency) { | ||
| this.logger.error({ | ||
| at: "PriceWorker", | ||
| message: "Unable to find base currency to quote", | ||
| ...params, | ||
| outputToken: relayHashInfo.fillEvent.outputToken, | ||
| destinationChainId: relayHashInfo.destinationChainId, | ||
| }); | ||
| return; | ||
| } | ||
| const existingPrice = await historicPriceRepository.findOne({ | ||
| where: { | ||
| date: priceTime, | ||
| baseCurrency, | ||
| quoteCurrency, | ||
| }, | ||
| }); | ||
| // fetch price if one hasnt been saved | ||
| if (!existingPrice) { | ||
| try { | ||
| const historicPriceData = | ||
| await this.coingeckoClient.getHistoricDailyPrice( | ||
| priceTime.getTime(), | ||
| baseCurrency, | ||
| ); | ||
| price = historicPriceData.market_data?.current_price[quoteCurrency]; | ||
| // wasnt able to get a price | ||
| if (price === undefined) { | ||
| this.logger.error( | ||
| `Unable to find ${quoteCurrency} for ${baseCurrency} at time ${priceTime}`, | ||
| ); | ||
| return; | ||
| } | ||
| await historicPriceRepository.insert({ | ||
| date: priceTime, | ||
| baseCurrency, | ||
| quoteCurrency, | ||
| price: price.toString(), | ||
| }); | ||
| this.logger.info({ | ||
| at: "PriceWorker", | ||
| ...params, | ||
| message: `Fetched and inserted historic price for ${baseCurrency} on ${priceTime}`, | ||
| }); | ||
| } catch (error) { | ||
| this.logger.error({ | ||
| at: "PriceWorker", | ||
| ...params, | ||
| message: `Failed to fetch or insert historic price for ${baseCurrency} on ${priceTime}`, | ||
| error: (error as Error).message, | ||
| }); | ||
| } | ||
| } else { | ||
| price = Number(existingPrice.price); | ||
| } | ||
|
|
||
| if (price === undefined) { | ||
| this.logger.error({ | ||
| at: "PriceWorker", | ||
| ...params, | ||
| message: "Failed to get a valid price from cache or coingecko", | ||
| }); | ||
| return; | ||
| } | ||
| // TODO: Compute bridge fee | ||
| } | ||
| public async close() { | ||
| return this.worker.close(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,3 @@ | ||
| export * from "./bundles"; | ||
| export * from "./spokePoolProcessor"; | ||
| export * from "./BundleBuilderService"; | ||
| export * from "./priceProcessor"; |
This file was deleted.
Oops, something went wrong.
Oops, something went wrong.