feat: assign spoke pool events to bundles (#77)

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

@@ -13,6 +13,7 @@ import { RootBundleCanceled } from "./evm/RootBundleCanceled";
 import { RootBundleExecuted } from "./evm/RootBundleExecuted";
 import { RootBundleDisputed } from "./evm/RootBundleDisputed";
 import { BundleBlockRange } from "./BundleBlockRange";
+import { BundleEvent } from "./BundleEvent";
 
 export enum BundleStatus {
   Proposed = "Proposed",
@@ -89,4 +90,10 @@ export class Bundle {
     nullable: false,
   })
   ranges: BundleBlockRange[];
+
+  @Column({ default: false })
+  eventsAssociated: boolean;
+
+  @OneToMany(() => BundleEvent, (event) => event.bundle)
+  events: BundleEvent[];
 }

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

@@ -0,0 +1,38 @@
+import {
+  Column,
+  Entity,
+  ManyToOne,
+  PrimaryGeneratedColumn,
+  Unique,
+} from "typeorm";
+import { Bundle } from "./Bundle";
+
+export enum BundleEventType {
+  Deposit = "deposit",
+  ExpiredDeposit = "expiredDeposit",
+  Fill = "fill",
+  SlowFill = "slowFill",
+  UnexecutableSlowFill = "unexecutableSlowFill",
+}
+
+@Entity()
+@Unique("UK_bundleEvent_eventType_relayHash", ["type", "relayHash"])
+export class BundleEvent {
+  @PrimaryGeneratedColumn()
+  id: number;
+
+  @ManyToOne(() => Bundle, (bundle) => bundle.events)
+  bundle: Bundle;
+
+  @Column()
+  bundleId: number;
+
+  @Column({ type: "enum", enum: BundleEventType })
+  type: BundleEventType;
+
+  @Column()
+  relayHash: string;
+
+  @Column({ nullable: true })
+  repaymentChainId: number;
+}

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

@@ -15,6 +15,7 @@ export * from "./evm/TokensBridged";
 
 // Others
 export * from "./Bundle";
+export * from "./BundleEvent";
 export * from "./BundleBlockRange";
 export * from "./RootBundleExecutedJoinTable";
 export * from "./RelayHashInfo";

packages/indexer-database/src/main.ts

@@ -39,6 +39,7 @@ export const createDataSource = (config: DatabaseConfig): DataSource => {
       // Bundle
       entities.Bundle,
       entities.BundleBlockRange,
+      entities.BundleEvent,
       entities.RootBundleExecutedJoinTable,
       // Others
       entities.RelayHashInfo,

packages/indexer-database/src/migrations/1729644581993-BundleEvent.ts

@@ -0,0 +1,38 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class BundleEvent1729644581993 implements MigrationInterface {
+  name = "BundleEvent1729644581993";
+
+  public async up(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `CREATE TYPE "public"."bundle_event_type_enum" AS ENUM('deposit', 'expiredDeposit', 'fill', 'slowFill', 'unexecutableSlowFill')`,
+    );
+    await queryRunner.query(
+      `CREATE TABLE "bundle_event" (
+        "id" SERIAL NOT NULL,
+        "bundleId" integer NOT NULL,
+        "type" "public"."bundle_event_type_enum" NOT NULL,
+        "relayHash" character varying NOT NULL,
+        "repaymentChainId" integer,
+        CONSTRAINT "UK_bundleEvent_eventType_relayHash" UNIQUE ("type", "relayHash"),
+        CONSTRAINT "PK_d633122fa4b52768e1b588bddee" PRIMARY KEY ("id"))`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "bundle" ADD "eventsAssociated" boolean NOT NULL DEFAULT false`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "bundle_event" ADD CONSTRAINT "FK_62dcd4f6f0d1713fab0c8542dba" FOREIGN KEY ("bundleId") REFERENCES "bundle"("id") ON DELETE NO ACTION ON UPDATE NO ACTION`,
+    );
+  }
+
+  public async down(queryRunner: QueryRunner): Promise<void> {
+    await queryRunner.query(
+      `ALTER TABLE "bundle_event" DROP CONSTRAINT "FK_62dcd4f6f0d1713fab0c8542dba"`,
+    );
+    await queryRunner.query(
+      `ALTER TABLE "bundle" DROP COLUMN "eventsAssociated"`,
+    );
+    await queryRunner.query(`DROP TABLE "bundle_event"`);
+    await queryRunner.query(`DROP TYPE "public"."bundle_event_type_enum"`);
+  }
+}

packages/indexer/src/database/BundleRepository.ts

@@ -1,4 +1,5 @@
 import winston from "winston";
+import * as across from "@across-protocol/sdk";
 import { DataSource, entities, utils } from "@repo/indexer-database";
 
 export type BlockRangeInsertType = {
@@ -32,6 +33,7 @@ export class BundleRepository extends utils.BaseRepository {
     postgres: DataSource,
     logger: winston.Logger,
     throwError?: boolean,
+    private chunkSize = 2000,
   ) {
     super(postgres, logger, throwError);
   }
@@ -371,4 +373,182 @@ export class BundleRepository extends utils.BaseRepository {
 
     return (await executedUpdateQuery.execute())?.affected ?? 0;
   }
+
+  /**
+   * Retrieves executed bundles that do not have events associated with them.
+   * The query can be filtered by the block number and a limit on the number of results returned.
+   * @param filters - Optional filters for the query.
+   * @param filters.fromBlock - If provided, retrieves bundles where the proposal's block number is greater than this value.
+   * @param limit - The maximum number of bundles to retrieve.
+   * @returns An array of bundles that match the given criteria.
+   */
+  public async getExecutedBundlesWithoutEventsAssociated(
+    filters: {
+      fromBlock?: number;
+    },
+    limit = 5,
+  ): Promise<entities.Bundle[]> {
+    const bundleRepo = this.postgres.getRepository(entities.Bundle);
+    const query = bundleRepo
+      .createQueryBuilder("b")
+      .select(["b", "proposal", "ranges"])
+      .leftJoinAndSelect("b.ranges", "ranges")
+      .leftJoinAndSelect("b.proposal", "proposal")
+      .where("b.status = :executed", {
+        executed: entities.BundleStatus.Executed,
+      })
+      .andWhere("b.eventsAssociated = false");
+    if (filters.fromBlock) {
+      query.andWhere("proposal.blockNumber > :fromBlock", {
+        fromBlock: filters.fromBlock,
+      });
+    }
+    return query.orderBy("proposal.blockNumber", "DESC").take(limit).getMany();
+  }
+
+  /**
+   * Updates the `eventsAssociated` flag to `true` for a specific bundle.
+   * @param bundleId - The ID of the bundle to update.
+   * @returns A promise that resolves when the update is complete.
+   */
+  public async updateBundleEventsAssociatedFlag(bundleId: number) {
+    const bundleRepo = this.postgres.getRepository(entities.Bundle);
+    const updatedBundle = await bundleRepo
+      .createQueryBuilder()
+      .update()
+      .set({ eventsAssociated: true })
+      .where("id = :id", { id: bundleId })
+      .execute();
+    return updatedBundle.affected;
+  }
+
+  /**
+   * Stores bundle events relating them to a given bundle.
+   * @param bundleData The reconstructed bundle data.
+   * @param bundleId ID of the bundle to associate these events with.
+   * @returns A promise that resolves when all the events have been inserted into the database.
+   */
+  public async storeBundleEvents(
+    bundleData: across.interfaces.LoadDataReturnValue,
+    bundleId: number,
+  ) {
+    const eventsRepo = this.postgres.getRepository(entities.BundleEvent);
+
+    // Store bundle deposits
+    const deposits = this.formatBundleEvents(
+      entities.BundleEventType.Deposit,
+      bundleData.bundleDepositsV3,
+      bundleId,
+    );
+    const chunkedDeposits = across.utils.chunk(deposits, this.chunkSize);
+    await Promise.all(
+      chunkedDeposits.map((eventsChunk) => eventsRepo.insert(eventsChunk)),
+    );
+
+    // Store bundle refunded deposits
+    const expiredDeposits = this.formatBundleEvents(
+      entities.BundleEventType.ExpiredDeposit,
+      bundleData.expiredDepositsToRefundV3,
+      bundleId,
+    );
+    const chunkedRefunds = across.utils.chunk(expiredDeposits, this.chunkSize);
+    await Promise.all(
+      chunkedRefunds.map((eventsChunk) => eventsRepo.insert(eventsChunk)),
+    );
+
+    // Store bundle slow fills
+    const slowFills = this.formatBundleEvents(
+      entities.BundleEventType.SlowFill,
+      bundleData.bundleSlowFillsV3,
+      bundleId,
+    );
+    const chunkedSlowFills = across.utils.chunk(slowFills, this.chunkSize);
+    await Promise.all(
+      chunkedSlowFills.map((eventsChunk) => eventsRepo.insert(eventsChunk)),
+    );
+
+    // Store bundle unexecutable slow fills
+    const unexecutableSlowFills = this.formatBundleEvents(
+      entities.BundleEventType.UnexecutableSlowFill,
+      bundleData.unexecutableSlowFills,
+      bundleId,
+    );
+    const chunkedUnexecutableSlowFills = across.utils.chunk(
+      unexecutableSlowFills,
+      this.chunkSize,
+    );
+    await Promise.all(
+      chunkedUnexecutableSlowFills.map((eventsChunk) =>
+        eventsRepo.insert(eventsChunk),
+      ),
+    );
+
+    // Store bundle fills
+    const fills = this.formatBundleFillEvents(
+      entities.BundleEventType.Fill,
+      bundleData.bundleFillsV3,
+      bundleId,
+    );
+    const chunkedFills = across.utils.chunk(fills, this.chunkSize);
+    await Promise.all(
+      chunkedFills.map((eventsChunk) => eventsRepo.insert(eventsChunk)),
+    );
+
+    return {
+      deposits: deposits.length,
+      expiredDeposits: expiredDeposits.length,
+      slowFills: slowFills.length,
+      unexecutableSlowFills: unexecutableSlowFills.length,
+      fills: fills.length,
+    };
+  }
+
+  private formatBundleEvents(
+    eventsType: entities.BundleEventType,
+    bundleEvents:
+      | across.interfaces.BundleDepositsV3
+      | across.interfaces.BundleSlowFills
+      | across.interfaces.BundleExcessSlowFills
+      | across.interfaces.ExpiredDepositsToRefundV3,
+    bundleId: number,
+  ): {
+    bundleId: number;
+    relayHash: string;
+    type: entities.BundleEventType;
+  }[] {
+    return Object.values(bundleEvents).flatMap((tokenEvents) =>
+      Object.values(tokenEvents).flatMap((events) =>
+        events.map((event) => {
+          return {
+            bundleId,
+            relayHash: across.utils.getRelayHashFromEvent(event),
+            type: eventsType,
+          };
+        }),
+      ),
+    );
+  }
+
+  private formatBundleFillEvents(
+    eventsType: entities.BundleEventType.Fill,
+    bundleEvents: across.interfaces.BundleFillsV3,
+    bundleId: number,
+  ): {
+    bundleId: number;
+    relayHash: string;
+    type: entities.BundleEventType.Fill;
+  }[] {
+    return Object.entries(bundleEvents).flatMap(([chainId, tokenEvents]) =>
+      Object.values(tokenEvents).flatMap((fillsData) =>
+        fillsData.fills.map((event) => {
+          return {
+            bundleId,
+            relayHash: across.utils.getRelayHashFromEvent(event),
+            type: eventsType,
+            repaymentChainId: Number(chainId),
+          };
+        }),
+      ),
+    );
+  }
 }

packages/indexer/src/main.ts

@@ -83,6 +83,8 @@ export async function Main(config: parseEnv.Config, logger: winston.Logger) {
     logger,
     redis,
     postgres,
+    hubPoolClientFactory,
+    spokePoolClientFactory,
   });
 
   const spokePoolIndexers = spokePoolChainsEnabled.map((chainId) => {

packages/indexer/src/services/BundleBuilderService.ts

@@ -9,6 +9,7 @@ import { BundleLeavesCache } from "../redis/bundleLeavesCache";
 import { HubPoolBalanceCache } from "../redis/hubBalancesCache";
 import {
   BN_ZERO,
+  buildPoolRebalanceRoot,
   ConfigStoreClientFactory,
   convertProposalRangeResultToProposalRange,
   getBlockRangeBetweenBundles,
@@ -398,30 +399,17 @@ export class BundleBuilderService extends BaseIndexer {
       chainsToBuildBundleFor,
     );
     // Load the bundle data
-    const {
-      bundleDepositsV3,
-      expiredDepositsToRefundV3,
-      bundleFillsV3,
-      unexecutableSlowFills,
-      bundleSlowFillsV3,
-    } = await bundleDataClient.loadData(
+    const bundleData = await bundleDataClient.loadData(
       bundleRangeForBundleClient,
       spokeClients,
       false,
     );
     // Build pool rebalance root and resolve the leaves
-    const { leaves } = clients.BundleDataClient._buildPoolRebalanceRoot(
-      bundleRangeForBundleClient[0]![1]!, // Mainnet is always the first chain. Second element is the end block
-      bundleRangeForBundleClient[0]![1]!, // Mainnet is always the first chain. Second element is the end block
-      bundleDepositsV3,
-      bundleFillsV3,
-      bundleSlowFillsV3,
-      unexecutableSlowFills,
-      expiredDepositsToRefundV3,
-      {
-        hubPoolClient,
-        configStoreClient,
-      },
+    const { leaves } = buildPoolRebalanceRoot(
+      bundleRangeForBundleClient,
+      bundleData,
+      hubPoolClient,
+      configStoreClient,
     );
     // Map the leaves to the desired format
     return leaves.map((leaf) => ({