use sdk coingecko, use floats for prices

Signed-off-by: david <david@umaproject.org>

packages/indexer-database/src/entities/HistoricPrice.ts

@@ -27,7 +27,7 @@ export class HistoricPrice {
   @Column({ type: "date" })
   date: Date;
 
-  @Column({ type: "decimal" })
+  @Column({ type: "float" })
   price: string;
 
   @CreateDateColumn()

packages/indexer-database/src/entities/RelayHashInfo.ts

@@ -92,12 +92,12 @@ export class RelayHashInfo {
   @CreateDateColumn()
   createdAt: Date;
 
-  @Column({ nullable: true })
+  @Column({ nullable: true, type: "float" })
   bridgeFeeUsd: string;
-  @Column({ nullable: true })
-  inputPriceUsd: number;
-  @Column({ nullable: true })
-  outputPriceUsd: number;
+  @Column({ nullable: true, type: "float" })
+  inputPriceUsd: string;
+  @Column({ nullable: true, type: "float" })
+  outputPriceUsd: string;
 
   @UpdateDateColumn()
   updatedAt: Date;

packages/indexer-database/src/migrations/1736274243965-RelayHashInfo.ts

@@ -0,0 +1,55 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class RelayHashInfo1736274243965 implements MigrationInterface {
+  name = "RelayHashInfo1736274243965";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" DROP COLUMN "bridgeFeeUsd"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" ADD "bridgeFeeUsd" double precision`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" DROP COLUMN "inputPriceUsd"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" ADD "inputPriceUsd" double precision`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" DROP COLUMN "outputPriceUsd"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" ADD "outputPriceUsd" double precision`,
+    );
+    await queryRunner.query(`ALTER TABLE "historic_price" DROP COLUMN "price"`);
+    await queryRunner.query(
+      `ALTER TABLE "historic_price" ADD "price" double precision NOT NULL`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`ALTER TABLE "historic_price" DROP COLUMN "price"`);
+    await queryRunner.query(
+      `ALTER TABLE "historic_price" ADD "price" numeric NOT NULL`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" DROP COLUMN "outputPriceUsd"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" ADD "outputPriceUsd" integer`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" DROP COLUMN "inputPriceUsd"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" ADD "inputPriceUsd" integer`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" DROP COLUMN "bridgeFeeUsd"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "relay_hash_info" ADD "bridgeFeeUsd" character varying`,
+    );
+  }
+}

packages/indexer/src/messaging/priceWorker.ts

@@ -1,19 +1,29 @@
 import Redis from "ioredis";
+import { DateTime } from "luxon";
 import winston from "winston";
 import { Job, Worker } from "bullmq";
 import { DataSource, entities } from "@repo/indexer-database";
 import { IndexerQueues } from "./service";
 import { ethers } from "ethers";
-import { yesterday, findTokenByAddress } from "../utils";
-import { CoingeckoClient } from "../utils/coingeckoClient";
+import { findTokenByAddress } from "../utils";
+// import { CoingeckoClient } from "../utils/coingeckoClient";
 import { RetryProvidersFactory } from "../web3/RetryProvidersFactory";
 import { assert } from "@repo/error-handling";
+import * as across from "@across-protocol/sdk";
 
 export type PriceMessage = {
   depositId: number;
   originChainId: number;
 };
 
+// Convert now to a consistent price timestamp yesterday for lookup purposes
+export function yesterday(now: Date) {
+  return DateTime.fromJSDate(now)
+    .minus({ days: 1 })
+    .set({ hour: 23, minute: 59, second: 0, millisecond: 0 })
+    .toJSDate();
+}
+
 /**
  * This worker listens to the `PriceQuery` queue and processes each job by:
  * - Retrieving the deposit and relay hash information from the database using the deposit ID and origin chain ID.
@@ -26,7 +36,7 @@ export type PriceMessage = {
  */
 export class PriceWorker {
   private worker: Worker;
-  private coingeckoClient: CoingeckoClient;
+  private coingeckoClient: across.coingecko.Coingecko;
   private relayHashInfoRepository;
   private depositRepository;
   private historicPriceRepository;
@@ -36,7 +46,7 @@ export class PriceWorker {
     private postgres: DataSource,
     private logger: winston.Logger,
   ) {
-    this.coingeckoClient = new CoingeckoClient();
+    this.coingeckoClient = across.coingecko.Coingecko.get(logger);
     this.relayHashInfoRepository = this.postgres.getRepository(
       entities.RelayHashInfo,
     );
@@ -66,14 +76,18 @@ export class PriceWorker {
       },
     });
     // we have this price at this time in the db
-    if (cachedPrice) return Number(cachedPrice.price);
+    if (cachedPrice) {
+      return Number(cachedPrice.price);
+    }
 
-    const fetchedPrice = await this.coingeckoClient.getHistoricDailyPrice(
-      priceTime.getTime(),
-      // use the coingecko id to fetch basecurrency price in usd
-      tokenInfo.coingeckoId,
+    const cgFormattedDate =
+      DateTime.fromJSDate(priceTime).toFormat("dd-LL-yyyy");
+    const price = await this.coingeckoClient.getContractHistoricDayPrice(
+      address,
+      cgFormattedDate,
+      quoteCurrency,
+      chainId,
     );
-    const price = fetchedPrice.market_data?.current_price[quoteCurrency];
     assert(
       price,
       `Unable to fetch price for ${quoteCurrency} in ${baseCurrency}(${tokenInfo.coingeckoId}) at ${priceTime}`,
@@ -160,6 +174,19 @@ export class PriceWorker {
       (info) => info.depositTxHash === (deposit && deposit.transactionHash),
     );
 
+    if (
+      relayHashInfo?.bridgeFeeUsd &&
+      relayHashInfo?.inputPriceUsd &&
+      relayHashInfo?.outputPriceUsd
+    ) {
+      const errorMessage = "Skipping already processed relay hash";
+      this.logger.error({
+        at: "PriceWorker",
+        message: errorMessage,
+        ...params,
+      });
+      return;
+    }
     const errorMessage =
       "Failed to retrieve relay hash information or deposit record from the database.";
 

packages/indexer/src/utils/index.ts

@@ -2,5 +2,4 @@ export * from "./contractUtils";
 export * from "./contractFactoryUtils";
 export * from "./bundleBuilderUtils";
 export * from "./spokePoolUtils";
-export * from "./coingeckoClient";
 export * from "./currencyUtils";