feat: send cctp burn transactions to pubsub

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

@@ -0,0 +1,42 @@
+import {
+  Column,
+  CreateDateColumn,
+  Entity,
+  JoinColumn,
+  OneToOne,
+  PrimaryGeneratedColumn,
+  Unique,
+  UpdateDateColumn,
+} from "typeorm";
+import { DepositForBurn } from "./evm/DepositForBurn";
+
+/**
+ * Table to store the burn events sent to be finalized by the
+ * finalizer bot. Each row points to a burn event that has
+ * been sent successfully to the finalizer bot.
+ */
+@Entity()
+@Unique("UK_cctpFinalizerJob_burnEventId", ["burnEventId"])
+export class CctpFinalizerJob {
+  @PrimaryGeneratedColumn()
+  id: number;
+
+  @Column()
+  attestation: string;
+
+  @Column()
+  message: string;
+
+  @Column()
+  burnEventId: number;
+
+  @OneToOne(() => DepositForBurn, (burnEvent) => burnEvent.id)
+  @JoinColumn({ name: "burnEventId" })
+  burnEvent: DepositForBurn;
+
+  @CreateDateColumn()
+  createdAt: Date;
+
+  @UpdateDateColumn()
+  updatedAt: Date;
+}

packages/indexer-database/src/entities/evm/DepositForBurn.ts

@@ -4,9 +4,11 @@ import {
   DeleteDateColumn,
   Entity,
   Index,
+  OneToOne,
   PrimaryGeneratedColumn,
   Unique,
 } from "typeorm";
+import { CctpFinalizerJob } from "../CctpFinalizerJob";
 
 @Entity({ schema: "evm" })
 @Unique("UK_depositForBurn_chain_block_txn_log", [
@@ -77,4 +79,7 @@ export class DepositForBurn {
 
   @DeleteDateColumn({ nullable: true })
   deletedAt?: Date;
+
+  @OneToOne(() => CctpFinalizerJob, (finalizerJob) => finalizerJob.burnEvent)
+  finalizerJob: CctpFinalizerJob;
 }

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

@@ -37,6 +37,8 @@ export * from "./evm/MintAndWithdraw";
 export * from "./evm/MessageReceived";
 export * from "./evm/SponsoredDepositForBurn";
 
+export * from "./CctpFinalizerJob";
+
 // OFT
 export * from "./evm/OftSent";
 export * from "./evm/OftReceived";

packages/indexer-database/src/main.ts

@@ -69,6 +69,7 @@ export const createDataSource = (config: DatabaseConfig): DataSource => {
       entities.MessageReceived,
       entities.MintAndWithdraw,
       entities.SponsoredDepositForBurn,
+      entities.CctpFinalizerJob,
       // OFT
       entities.OFTSent,
       entities.OFTReceived,

packages/indexer-database/src/migrations/1760541259411-CctpFinalizerJob.ts

@@ -0,0 +1,29 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class CctpFinalizerJob1760541259411 implements MigrationInterface {
+  name = "CctpFinalizerJob1760541259411";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(`
+      CREATE TABLE "cctp_finalizer_job" (
+        "id" SERIAL NOT NULL,
+        "attestation" character varying NOT NULL,
+        "message" character varying NOT NULL,
+        "burnEventId" integer NOT NULL,
+        "createdAt" TIMESTAMP NOT NULL DEFAULT now(),
+        "updatedAt" TIMESTAMP NOT NULL DEFAULT now(),
+        CONSTRAINT "REL_955f0780d44fb1b785425d7b56" UNIQUE ("burnEventId"),
+        CONSTRAINT "PK_f6a74f608a35916f58fbb1da6ba" PRIMARY KEY ("id")
+      )`);
+    await queryRunner.query(
+      `ALTER TABLE "cctp_finalizer_job" ADD CONSTRAINT "FK_955f0780d44fb1b785425d7b567" FOREIGN KEY ("burnEventId") REFERENCES "evm"."deposit_for_burn"("id") ON DELETE NO ACTION ON UPDATE NO ACTION`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "cctp_finalizer_job" DROP CONSTRAINT "FK_955f0780d44fb1b785425d7b567"`,
+    );
+    await queryRunner.query(`DROP TABLE "cctp_finalizer_job"`);
+  }
+}

packages/indexer/README.md

@@ -53,6 +53,10 @@ ENABLE_BUNDLE_INCLUDED_EVENTS_SERVICE=true
 ENABLE_BUNDLE_BUILDER=true
 ENABLE_PRICE_WORKER=false
 ENABLE_CCTP_INDEXER=false
+ENABLE_CCTP_FINALIZER=false
+
+PUBSUB_CCTP_FINALIZER_TOPIC=
+PUBSUB_GCP_PROJECT_ID=
 
 COINGECKO_API_KEY=
 BUNDLE_EVENTS_SERVICE_START_BLOCK_NUMBER=

packages/indexer/package.json

@@ -24,12 +24,14 @@
     "@across-protocol/constants": "^3.1.82",
     "@across-protocol/contracts": "^4.1.11",
     "@across-protocol/sdk": "^4.3.75",
+    "@google-cloud/pubsub": "^5.2.0",
     "@repo/error-handling": "workspace:*",
     "@repo/webhooks": "workspace:*",
     "@solana/kit": "^2.1.0",
     "@types/express": "^4.17.21",
     "@types/lodash": "^4.17.7",
     "@types/luxon": "^3.4.2",
+    "axios": "^1.12.2",
     "bullmq": "^5.12.12",
     "ethers": "^5.8.0",
     "express": "^4.19.2",

packages/indexer/src/data-indexing/adapter/cctp-v2/service.ts

@@ -1,8 +1,10 @@
 import { ethers } from "ethers";
+import axios from "axios";
 import {
   CCTP_NO_DOMAIN,
   CHAIN_IDs,
   PRODUCTION_NETWORKS,
+  PUBLIC_NETWORKS,
   TEST_NETWORKS,
 } from "@across-protocol/constants";
 import * as across from "@across-protocol/sdk";
@@ -81,6 +83,19 @@ export function decodeMessage(messageBytesArray: Uint8Array) {
   };
 }
 
+/**
+ * @notice Returns the CCTP domain for a given chain ID. Throws if the chain ID is not a CCTP domain.
+ * @param chainId
+ * @returns CCTP Domain ID
+ */
+export function getCctpDomainForChainId(chainId: number): number {
+  const cctpDomain = PUBLIC_NETWORKS[chainId]?.cctpDomain;
+  if (!across.utils.isDefined(cctpDomain) || cctpDomain === CCTP_NO_DOMAIN) {
+    throw new Error(`No CCTP domain found for chainId: ${chainId}`);
+  }
+  return cctpDomain;
+}
+
 export function getCctpDestinationChainFromDomain(
   domain: number,
   productionNetworks: boolean = true,
@@ -111,3 +126,27 @@ export function getCctpDestinationChainFromDomain(
   }
   return parseInt(chainId);
 }
+
+export type CCTPV2APIGetAttestationResponse = {
+  messages: CCTPV2APIAttestation[];
+};
+
+export type CCTPV2APIAttestation = {
+  eventNonce: string;
+  status: string;
+  attestation: string;
+  message: string;
+};
+
+export async function fetchAttestationsForTxn(
+  sourceDomainId: number,
+  transactionHash: string,
+  isMainnet: boolean,
+): Promise<CCTPV2APIGetAttestationResponse> {
+  const httpResponse = await axios.get<CCTPV2APIGetAttestationResponse>(
+    `https://iris-api${
+      isMainnet ? "" : "-sandbox"
+    }.circle.com/v2/messages/${sourceDomainId}?transactionHash=${transactionHash}`,
+  );
+  return httpResponse.data;
+}

packages/indexer/src/data-indexing/service/CCTPIndexerManager.ts

@@ -74,7 +74,8 @@ export class CCTPIndexerManager {
       );
       const indexer = new EvmIndexer(
         {
-          indexingDelaySeconds: getIndexingDelaySeconds(chainId, this.config),
+          indexingDelaySeconds:
+            getIndexingDelaySeconds(chainId, this.config) * 2,
           finalisedBlockBufferDistance:
             getFinalisedBlockBufferDistance(chainId),
           maxBlockRangeSize: MAX_BLOCK_RANGE_SIZE,