use relay hash to find entities

Signed-off-by: david <david@umaproject.org>
across-protocol · Jan 13, 2025 · c245020 · c245020
1 parent ca4498f
commit c245020
Show file tree

Hide file tree

Showing 5 changed files with 60 additions and 37 deletions.
diff --git 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
@@ -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
@@ -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
@@ -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<entities.FilledV3Relay>[],
+  ) {
+    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
@@ -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() {