Integrate pubsub lib

Author: amateima

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

@@ -24,6 +24,9 @@ export class CctpFinalizerJob {
   @Column()
   attestation: string;
 
+  @Column()
+  message: string;
+
   @Column()
   burnEventId: number;
 

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

@@ -8,6 +8,7 @@ export class CctpFinalizerJob1760541259411 implements MigrationInterface {
       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(),

packages/indexer/package.json

@@ -24,6 +24,7 @@
     "@across-protocol/constants": "^3.1.79",
     "@across-protocol/contracts": "^4.1.9",
     "@across-protocol/sdk": "^4.3.67",
+    "@google-cloud/pubsub": "^5.2.0",
     "@repo/error-handling": "workspace:*",
     "@repo/webhooks": "workspace:*",
     "@solana/kit": "^2.1.0",

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(message: string, isSvm = false) {
   };
 }
 
+/**
+ * @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

@@ -14,7 +14,7 @@ import { RetryProvidersFactory } from "../../web3/RetryProvidersFactory";
 import { CCTPIndexerDataHandler } from "./CCTPIndexerDataHandler";
 import { CCTPRepository } from "../../database/CctpRepository";
 
-const MAX_BLOCK_RANGE_SIZE = 10000;
+const MAX_BLOCK_RANGE_SIZE = 30000;
 
 export class CCTPIndexerManager {
   private evmIndexer?: Indexer;
@@ -69,7 +69,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,

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

@@ -2,14 +2,19 @@ import winston, { Logger } from "winston";
 import axios from "axios";
 
 import { RepeatableTask } from "../../generics";
-import { DataSource } from "@repo/indexer-database";
+import { DataSource, entities } from "@repo/indexer-database";
 import {
   CctpFinalizerJob,
   DepositForBurn,
 } from "../../../../indexer-database/dist/src/entities";
 import { CHAIN_IDs } from "@across-protocol/constants";
 import { PubSubService } from "../../pubsub/service";
 import { Config } from "../../parseEnv";
+import {
+  fetchAttestationsForTxn,
+  getCctpDestinationChainFromDomain,
+  getCctpDomainForChainId,
+} from "../adapter/cctp-v2/service";
 
 export const CCTP_FINALIZER_DELAY_SECONDS = 10;
 
@@ -136,6 +141,43 @@ class CctpFinalizerService extends RepeatableTask {
         });
         return;
       }
+      const attestations = await fetchAttestationsForTxn(
+        getCctpDomainForChainId(Number(burnEvent.chainId)),
+        transactionHash,
+        true,
+      );
+      if (attestations.messages.length === 0) {
+        this.logger.debug({
+          at: "CctpFinalizerService#publishBurnEvent",
+          message: "No attestations found for burn event",
+          chainId,
+          transactionHash,
+          burnEvent,
+        });
+        return;
+      }
+      const { attestation, eventNonce, message, status } =
+        attestations.messages[0]!;
+      if (status !== "complete") {
+        this.logger.debug({
+          at: "CctpFinalizerService#publishBurnEvent",
+          message: "Attestation is not complete",
+          chainId,
+          transactionHash,
+          burnEvent,
+          attestations,
+        });
+        return;
+      }
+
+      await this.postgres
+        .createQueryBuilder(entities.MessageSent, "ms")
+        .update()
+        .set({
+          nonce: eventNonce,
+        })
+        .where("id = :id", { id: burnEvent.id })
+        .execute();
       this.logger.debug({
         at: "CctpFinalizerService#publishBurnEvent",
         message: "Publishing burn event to pubsub",
@@ -146,16 +188,23 @@ class CctpFinalizerService extends RepeatableTask {
         attestationTimeSeconds,
         elapsedSeconds,
       });
+      const destinationChainId = getCctpDestinationChainFromDomain(
+        burnEvent.destinationDomain,
+      );
       await this.pubSubService.publishCctpFinalizerMessage(
         transactionHash,
         Number(chainId),
+        message,
+        attestation,
+        destinationChainId,
       );
 
       await this.postgres
         .createQueryBuilder(CctpFinalizerJob, "j")
         .insert()
         .values({
-          attestation: "",
+          attestation,
+          message,
           burnEventId: burnEvent.id,
         })
         .orUpdate(["attestation"], ["burnEventId"])

packages/indexer/src/parseEnv.ts

@@ -21,7 +21,8 @@ export type Config = {
   enableBundleBuilder: boolean;
   enableCctpIndexer: boolean;
   enableCctpFinalizer: boolean;
-  finalizerPubSubTopic: string;
+  pubSubCctpFinalizerTopic: string;
+  pubSubGcpProjectId: string;
   enableOftIndexer: boolean;
   webhookConfig: WebhooksConfig;
   maxBlockRangeSize?: number;
@@ -226,7 +227,8 @@ export function envToConfig(env: Env): Config {
   const enableCctpFinalizer = env.ENABLE_CCTP_FINALIZER
     ? env.ENABLE_CCTP_FINALIZER === "true"
     : false;
-  const finalizerPubSubTopic = env.CCTP_FINALIZER_PUBSUB_TOPIC ?? "";
+  const pubSubCctpFinalizerTopic = env.PUBSUB_CCTP_FINALIZER_TOPIC ?? "";
+  const pubSubGcpProjectId = env.PUBSUB_GCP_PROJECT_ID ?? "";
   const enableBundleIncludedEventsService =
     env.ENABLE_BUNDLE_INCLUDED_EVENTS_SERVICE
       ? env.ENABLE_BUNDLE_INCLUDED_EVENTS_SERVICE === "true"
@@ -286,7 +288,8 @@ export function envToConfig(env: Env): Config {
     enableCctpIndexer,
     enableOftIndexer,
     enableCctpFinalizer,
-    finalizerPubSubTopic,
+    pubSubCctpFinalizerTopic,
+    pubSubGcpProjectId,
     webhookConfig,
     maxBlockRangeSize,
     coingeckoApiKey,

packages/indexer/src/pubsub/service.ts

@@ -1,34 +1,43 @@
-import axios from "axios";
+import { PubSub, Topic } from "@google-cloud/pubsub";
 import { Config } from "../parseEnv";
 
 /**
  * Helper class to publish messages to a GCP pubsub topic.
  */
 export class PubSubService {
-  constructor(private readonly config: Config) {}
+  private readonly pubSub: PubSub;
+  private cctpFinalizerTopic: Topic;
+
+  constructor(private readonly config: Config) {
+    this.pubSub = new PubSub({
+      projectId: this.config.pubSubGcpProjectId,
+    });
+  }
 
   async publishCctpFinalizerMessage(
     burnTransactionHash: string,
     sourceChainId: number,
+    message: string,
+    attestation: string,
+    destinationChainId: number,
   ) {
+    if (!this.cctpFinalizerTopic) {
+      const topic = await this.pubSub.topic(
+        this.config.pubSubCctpFinalizerTopic,
+      );
+      this.cctpFinalizerTopic = topic;
+    }
     // the published payload is a base64 encoded JSON string. The JSON is
     // validated by the Avro schema defined in GCP
     const payload = Buffer.from(
       JSON.stringify({
         burnTransactionHash: burnTransactionHash,
         sourceChainId,
+        message,
+        attestation,
+        destinationChainId,
       }),
-    ).toString("base64");
-    const body = { messages: [{ data: payload }] };
-    const response = await axios.post(this.config.finalizerPubSubTopic, body, {
-      headers: {
-        "Content-Type": "application/json",
-        // TODO: this authorization method is temporary for local testing.
-        // It must be replaced with a proper authentication method.
-        Authorization: `Bearer <AUTH_TOKEN>`,
-      },
-    });
-
-    return response.data;
+    );
+    await this.cctpFinalizerTopic.publishMessage({ data: payload });
   }
 }