diff --git a/packages/indexer-database/src/entities/HistoricPrice.ts b/packages/indexer-database/src/entities/HistoricPrice.ts index 22c00a8..f35ad61 100644 --- a/packages/indexer-database/src/entities/HistoricPrice.ts +++ b/packages/indexer-database/src/entities/HistoricPrice.ts @@ -16,18 +16,16 @@ export class HistoricPrice { @PrimaryGeneratedColumn() id: number; - // bear in mind we are using coingecko symbols directly here, for all intents and purposes this is coingecko historic market price @Column() baseCurrency: string; @Column({ default: "usd" }) quoteCurrency: string; - // yyyy-LL-dd @Column({ type: "date" }) date: Date; - @Column({ type: "float" }) + @Column({ type: "decimal" }) price: string; @CreateDateColumn() diff --git a/packages/indexer-database/src/entities/RelayHashInfo.ts b/packages/indexer-database/src/entities/RelayHashInfo.ts index 5cca566..ce781e0 100644 --- a/packages/indexer-database/src/entities/RelayHashInfo.ts +++ b/packages/indexer-database/src/entities/RelayHashInfo.ts @@ -92,11 +92,11 @@ export class RelayHashInfo { @CreateDateColumn() createdAt: Date; - @Column({ nullable: true, type: "float" }) + @Column({ nullable: true, type: "decimal" }) bridgeFeeUsd: string; - @Column({ nullable: true, type: "float" }) + @Column({ nullable: true, type: "decimal" }) inputPriceUsd: string; - @Column({ nullable: true, type: "float" }) + @Column({ nullable: true, type: "decimal" }) outputPriceUsd: string; @UpdateDateColumn() diff --git a/packages/indexer-database/src/main.ts b/packages/indexer-database/src/main.ts index 96c9bf4..cbae5ea 100644 --- a/packages/indexer-database/src/main.ts +++ b/packages/indexer-database/src/main.ts @@ -39,6 +39,8 @@ export const createDataSource = (config: DatabaseConfig): DataSource => { // Webhooks entities.WebhookRequest, entities.WebhookClient, + // Historic Price + entities.HistoricPrice, ], migrationsTableName: "_migrations", migrations: ["migrations/*.ts"], diff --git a/packages/indexer/src/data-indexing/service/SpokePoolIndexerDataHandler.ts b/packages/indexer/src/data-indexing/service/SpokePoolIndexerDataHandler.ts index 36d93a5..003b73c 100644 --- a/packages/indexer/src/data-indexing/service/SpokePoolIndexerDataHandler.ts +++ b/packages/indexer/src/data-indexing/service/SpokePoolIndexerDataHandler.ts @@ -118,7 +118,7 @@ export class SpokePoolIndexerDataHandler implements IndexerDataHandler { await this.spokePoolProcessor.process(storedEvents); this.profileStoreEvents(storedEvents); // publish new relays to workers to fill in prices - await this.publishNewRelays(events.filledV3RelayEvents); + await this.publishNewRelays(storedEvents.fills); } /** @@ -336,13 +336,19 @@ export class SpokePoolIndexerDataHandler implements IndexerDataHandler { }); } - private async publishNewRelays(relays: FillWithBlock[]) { - const messages: PriceMessage[] = relays.map((relay) => { - return { - depositId: relay.depositId, - originChainId: relay.originChainId, - }; - }); + private async publishNewRelays( + fills: SaveQueryResult[], + ) { + const messages: PriceMessage[] = fills + .map((fill) => ({ + relayHash: fill.data?.relayHash, + originChainId: fill.data?.originChainId, + })) + .filter( + (x): x is PriceMessage => + x.relayHash !== undefined && x.originChainId !== undefined, + ); + await this.indexerQueuesService.publishMessagesBulk( IndexerQueues.PriceQuery, IndexerQueues.PriceQuery, // Use queue name as job name diff --git a/packages/indexer/src/messaging/priceWorker.ts b/packages/indexer/src/messaging/priceWorker.ts index e2655e8..d14beeb 100644 --- a/packages/indexer/src/messaging/priceWorker.ts +++ b/packages/indexer/src/messaging/priceWorker.ts @@ -12,7 +12,7 @@ import { assert } from "@repo/error-handling"; import * as across from "@across-protocol/sdk"; export type PriceMessage = { - depositId: number; + relayHash: string; originChainId: number; }; @@ -111,7 +111,7 @@ export class PriceWorker { } catch (error) { this.logger.error({ at: "PriceWorker", - message: `Error getting price for deposit ${job.data.depositId} on chain ${job.data.originChainId}`, + message: `Error getting price for deposit ${job.data.relayHash}`, error, }); throw error; @@ -160,48 +160,59 @@ export class PriceWorker { ); } private async run(params: PriceMessage) { - const { depositId, originChainId } = params; + const { relayHash, originChainId } = params; - const relayHashInfoArray = await this.relayHashInfoRepository.find({ - where: { depositId, originChainId }, + const relayHashInfo = await this.relayHashInfoRepository.findOne({ + where: { relayHash, originChainId }, + relations: { + depositEvent: true, + fillEvent: true, + }, }); + + if (!relayHashInfo) { + const errorMessage = `Relay hash info not found by relay hash ${relayHash}`; + this.logger.error({ + at: "PriceWorker", + message: errorMessage, + ...params, + }); + throw new Error(errorMessage); + } + + const { depositId } = relayHashInfo; const deposit = await this.depositRepository.findOne({ where: { depositId, originChainId }, }); - // if we have multiple relays for same deposit, find hte one which matches the deposit hash - // the others would be invalid fills - const relayHashInfo = relayHashInfoArray.find( - (info) => info.depositTxHash === (deposit && deposit.transactionHash), - ); - if ( - relayHashInfo?.bridgeFeeUsd && - relayHashInfo?.inputPriceUsd && - relayHashInfo?.outputPriceUsd - ) { - const errorMessage = "Skipping already processed relay hash"; + if (!deposit) { + const errorMessage = `Unable to find deposit ${depositId} on chain ${originChainId}`; this.logger.error({ at: "PriceWorker", message: errorMessage, ...params, }); - return; + throw new Error(errorMessage); } - const errorMessage = - "Failed to retrieve relay hash information or deposit record from the database."; - // we will keep retrying until found or we know there was a reorg - if (!relayHashInfo || !deposit) { + if ( + relayHashInfo.bridgeFeeUsd && + relayHashInfo.inputPriceUsd && + relayHashInfo.outputPriceUsd + ) { + const errorMessage = "Skipping already processed relay hash"; this.logger.error({ at: "PriceWorker", message: errorMessage, ...params, }); - throw new Error(errorMessage); + return; } + const errorMessage = + "Failed to retrieve relay hash information or deposit record from the database."; // if blockTimestamp doesnt exist, maybe we keep retrying till it does - const blockTime = relayHashInfo?.depositEvent?.blockTimestamp; + const blockTime = relayHashInfo.depositEvent.blockTimestamp; if (!blockTime) { const errorMessage = "Deposit block time not found for relay hash info."; this.logger.error({ @@ -261,6 +272,12 @@ export class PriceWorker { { depositId, originChainId }, updatedFields, ); + this.logger.info({ + at: "PriceWorker#updateRelayHashInfo", + message: "Updated relay hash info with new fields", + params, + updatedFields, + }); } } public async close() {