-
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.
feat: fetch integrator id asynchronously when backfilling (#83)
- Loading branch information
1 parent
58fda7f
commit 323be1c
Showing
8 changed files
with
191 additions
and
40 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| 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 } from "../utils"; | ||
| import { RetryProvidersFactory } from "../web3/RetryProvidersFactory"; | ||
|
|
||
| export type IntegratorIdMessage = { | ||
| relayHash: string; | ||
| }; | ||
|
|
||
| /** | ||
| * 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 IntegratorIdWorker { | ||
| public worker: Worker; | ||
| constructor( | ||
| private redis: Redis, | ||
| private postgres: DataSource, | ||
| private logger: winston.Logger, | ||
| private providerFactory: RetryProvidersFactory, | ||
| ) { | ||
| this.setWorker(); | ||
| } | ||
|
|
||
| public setWorker() { | ||
| this.worker = new Worker( | ||
| IndexerQueues.IntegratorId, | ||
| async (job: Job<IntegratorIdMessage>) => { | ||
| const { relayHash } = job.data; | ||
| const repository = this.postgres.getRepository( | ||
| entities.V3FundsDeposited, | ||
| ); | ||
| const deposit = await repository.findOne({ | ||
| where: { relayHash }, | ||
| }); | ||
| if (!deposit) { | ||
| this.logger.warn({ | ||
| at: "IntegratorIdWorker", | ||
| message: `Skipping deposit with relay hash ${relayHash}. Not found in the database.`, | ||
| }); | ||
| return; | ||
| } | ||
| if (deposit.integratorId !== null) { | ||
| this.logger.info({ | ||
| at: "IntegratorIdWorker", | ||
| message: `Skipping deposit with relay hash ${relayHash}. IntegratorId field already populated.`, | ||
| }); | ||
| return; | ||
| } | ||
| const provider = this.providerFactory.getProviderForChainId( | ||
| deposit.originChainId, | ||
| ); | ||
| const integratorId = await getIntegratorId( | ||
| provider, | ||
| deposit.quoteTimestamp, | ||
| deposit.transactionHash, | ||
| ); | ||
| if (integratorId) { | ||
| await repository.update({ relayHash }, { integratorId }); | ||
| } | ||
| return; | ||
| }, | ||
| { connection: this.redis, concurrency: 10 }, | ||
| ); | ||
| } | ||
| } |
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 |
|---|---|---|
| @@ -1,3 +1,4 @@ | ||
| export * from "./contractUtils"; | ||
| export * from "./contractFactoryUtils"; | ||
| export * from "./bundleBuilderUtils"; | ||
| export * from "./spokePoolUtils"; |
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,39 @@ | ||
| import { interfaces, providers } from "@across-protocol/sdk"; | ||
|
|
||
| export type V3FundsDepositedWithIntegradorId = interfaces.DepositWithBlock & { | ||
| integratorId?: string | undefined; | ||
| }; | ||
|
|
||
| /** | ||
| * Retrieves the 4-character integrator ID from the transaction data | ||
| * associated with the provided transaction hash, if present. | ||
| * The integrator ID is expected to be found after the delimiter "1dc0de" in the transaction data. | ||
| * @async | ||
| * @param provider The provider to fetch transaction details from. | ||
| * @param depositDate | ||
| * @param txHash The transaction hash to retrieve the input data of. | ||
| * @returns The 4-character integrator ID if found, otherwise undefined. | ||
| */ | ||
| export async function getIntegratorId( | ||
| provider: providers.RetryProvider, | ||
| depositDate: Date, | ||
| txHash: string, | ||
| ) { | ||
| // If deposit was made before integratorId implementation, skip request | ||
| const INTEGRATOR_ID_IMPLEMENTATION_DATE = new Date(1718274000 * 1000); | ||
| if (depositDate < INTEGRATOR_ID_IMPLEMENTATION_DATE) { | ||
| return; | ||
| } | ||
| const INTEGRATOR_DELIMITER = "1dc0de"; | ||
| const INTEGRATOR_ID_LENGTH = 4; // Integrator ids are 4 characters long | ||
| let integratorId = undefined; | ||
| const txn = await provider.getTransaction(txHash); | ||
| const txnData = txn.data; | ||
| if (txnData.includes(INTEGRATOR_DELIMITER)) { | ||
| integratorId = txnData | ||
| .split(INTEGRATOR_DELIMITER) | ||
| .pop() | ||
| ?.substring(0, INTEGRATOR_ID_LENGTH); | ||
| } | ||
| return integratorId; | ||
| } |