fix: replace bulk upserts with inserts/updates if needed

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

@@ -70,7 +70,7 @@ export class V3FundsDeposited {
   quoteBlockNumber: number;
 
   @Column({ nullable: true })
-  integratorId: string;
+  integratorId?: string;
 
   @Column()
   transactionHash: string;

packages/indexer-database/src/index.ts

@@ -1,3 +1,4 @@
 export * from "./main";
 export * as entities from "./entities";
 export * as utils from "./utils";
+export * from "./model";

packages/indexer-database/src/main.ts

@@ -1,15 +1,9 @@
 import "reflect-metadata";
 import { DataSource, LessThan, Not, In } from "typeorm";
 import * as entities from "./entities";
+import { DatabaseConfig } from "./model";
 
 export { DataSource, LessThan, Not, In };
-export type DatabaseConfig = {
-  host: string;
-  port: string;
-  user: string;
-  password: string;
-  dbName: string;
-};
 
 export const createDataSource = (config: DatabaseConfig): DataSource => {
   return new DataSource({

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

@@ -0,0 +1,28 @@
+export type DatabaseConfig = {
+  host: string;
+  port: string;
+  user: string;
+  password: string;
+  dbName: string;
+};
+
+/**
+ * Enum to represent the result type of a query.
+ * - If the entity is identical to the one in the database, return `Nothing`.
+ * - If the unique keys are not present, return Inserted.
+ * - If the finalised field was the only one that changed, return `Finalised`.
+ * - If any of the entity fields were changed, return Updated.
+ * - If both the finalised field and other fields were changed, return UpdatedAndFinalised.
+ */
+export enum SaveQueryResultType {
+  Nothing = "nothing",
+  Inserted = "inserted",
+  Finalised = "finalised",
+  Updated = "updated",
+  UpdatedAndFinalised = "updatedAndFinalised",
+}
+
+export type SaveQueryResult<T> = {
+  data: T | undefined;
+  result: SaveQueryResultType;
+};

packages/indexer-database/src/utils/BlockchainEventRepository.ts

@@ -0,0 +1,114 @@
+import { DataSource, EntityTarget, ObjectLiteral } from "typeorm";
+import winston from "winston";
+
+import { SaveQueryResultType, SaveQueryResult } from "../model";
+
+export function filterSaveQueryResults<Entity extends ObjectLiteral>(
+  results: SaveQueryResult<Entity>[],
+  type: SaveQueryResultType,
+) {
+  return results
+    .filter((result) => result.result === type)
+    .map((result) => result.data)
+    .filter((data) => data !== undefined);
+}
+
+export class BlockchainEventRepository {
+  constructor(
+    protected postgres: DataSource,
+    protected logger: winston.Logger,
+  ) {}
+
+  /**
+   * Saves the entities to the database.
+   * @param entity - The entity to save.
+   * @param data - The data to save.
+   * @param uniqueKeys
+   * The unique keys to check for. It is recommended these keys to be indexed columns, so that the query is faster.
+   * @param comparisonKeys - The keys to compare for changes.
+   */
+  protected async saveAndHandleFinalisationBatch<Entity extends ObjectLiteral>(
+    entity: EntityTarget<Entity>,
+    data: Partial<Entity>[],
+    uniqueKeys: (keyof Entity)[],
+    comparisonKeys: (keyof Entity)[],
+  ): Promise<SaveQueryResult<Entity>[]> {
+    return Promise.all(
+      data.map((dataItem) =>
+        this.saveAndHandleFinalisation(
+          entity,
+          dataItem,
+          uniqueKeys,
+          comparisonKeys,
+        ),
+      ),
+    );
+  }
+
+  /**
+   * Saves the entity to the database.
+   * @param entity - The entity to save.
+   * @param data - The data to save.
+   * @param uniqueKeys
+   * The unique keys to check for. It is recommended these keys to be indexed columns, so that the query is faster.
+   * @param comparisonKeys - The keys to compare for changes.
+   */
+  protected async saveAndHandleFinalisation<Entity extends ObjectLiteral>(
+    entity: EntityTarget<Entity>,
+    data: Partial<Entity>,
+    uniqueKeys: (keyof Entity)[],
+    comparisonKeys: (keyof Entity)[],
+  ): Promise<SaveQueryResult<Entity>> {
+    const where = uniqueKeys.reduce(
+      (acc, key) => {
+        acc[key] = data[key];
+        return acc;
+      },
+      {} as Record<keyof Entity, any>,
+    );
+    const dbEntity = await this.postgres
+      .getRepository(entity)
+      .findOne({ where });
+    const repository = this.postgres.getRepository(entity);
+
+    if (!dbEntity) {
+      await repository.insert(data);
+      return {
+        data: (await repository.findOne({ where })) as Entity,
+        result: SaveQueryResultType.Inserted,
+      };
+    }
+
+    // Check if any of the values of the comparison keys have changed
+    const isChanged = comparisonKeys.some((key) => data[key] !== dbEntity[key]);
+    // Check if the data moved in finalised state
+    const isFinalisedChanged = data.finalised && !dbEntity.finalised;
+
+    if (isChanged) {
+      await repository.update(where, data);
+      if (isFinalisedChanged) {
+        return {
+          data: (await repository.findOne({ where })) as Entity,
+          result: SaveQueryResultType.UpdatedAndFinalised,
+        };
+      }
+      return {
+        data: (await repository.findOne({ where })) as Entity,
+        result: SaveQueryResultType.Updated,
+      };
+    }
+
+    if (isFinalisedChanged) {
+      await repository.update(where, data);
+      return {
+        data: (await repository.findOne({ where })) as Entity,
+        result: SaveQueryResultType.Finalised,
+      };
+    }
+
+    return {
+      data: undefined,
+      result: SaveQueryResultType.Nothing,
+    };
+  }
+}

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

@@ -1 +1,2 @@
 export * from "./BaseRepository";
+export * from "./BlockchainEventRepository";

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

@@ -110,12 +110,12 @@ export class AcrossIndexerManager {
         return spokePoolIndexer;
       },
     );
+    this.spokePoolIndexers = spokePoolIndexers;
 
-    if (spokePoolIndexers.length === 0) {
+    if (this.spokePoolIndexers.length === 0) {
       this.logger.warn("No spoke pool indexers to start");
       return;
     }
-    this.spokePoolIndexers = spokePoolIndexers;
     return Promise.all(
       this.spokePoolIndexers.map((indexer) => indexer.start()),
     );

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

@@ -4,18 +4,24 @@ import {
   getDeployedAddress,
   getDeployedBlockNumber,
 } from "@across-protocol/contracts";
-import { entities } from "@repo/indexer-database";
+import {
+  entities,
+  utils as indexerDatabaseUtils,
+  SaveQueryResult,
+} from "@repo/indexer-database";
+import { SaveQueryResultType } from "@repo/indexer-database";
 
 import { BlockRange } from "../model";
 import { IndexerDataHandler } from "./IndexerDataHandler";
 
 import * as utils from "../../utils";
+import { getIntegratorId } from "../../utils/spokePoolUtils";
 import { SpokePoolRepository } from "../../database/SpokePoolRepository";
 import { SpokePoolProcessor } from "../../services/spokePoolProcessor";
 import { IndexerQueues, IndexerQueuesService } from "../../messaging/service";
 import { IntegratorIdMessage } from "../../messaging/IntegratorIdWorker";
 
-type FetchEventsResult = {
+export type FetchEventsResult = {
   v3FundsDepositedEvents: utils.V3FundsDepositedWithIntegradorId[];
   filledV3RelayEvents: across.interfaces.FillWithBlock[];
   requestedV3SlowFillEvents: across.interfaces.SlowFillRequestWithBlock[];
@@ -29,6 +35,13 @@ type FetchEventsResult = {
   tokensBridgedEvents: across.interfaces.TokensBridged[];
 };
 
+export type StoreEventsResult = {
+  deposits: SaveQueryResult<entities.V3FundsDeposited>[];
+  fills: SaveQueryResult<entities.FilledV3Relay>[];
+  slowFillRequests: SaveQueryResult<entities.RequestedV3SlowFill>[];
+  executedRefundRoots: SaveQueryResult<entities.ExecutedRelayerRefundRoot>[];
+};
+
 export class SpokePoolIndexerDataHandler implements IndexerDataHandler {
   private isInitialized: boolean;
   private configStoreClient: across.clients.AcrossConfigStoreClient;
@@ -95,20 +108,12 @@ export class SpokePoolIndexerDataHandler implements IndexerDataHandler {
       blockRange,
       identifier: this.getDataIdentifier(),
     });
-
-    // Fetch integratorId synchronously when there are fewer than 1K deposit events
-    // For larger sets, use the IntegratorId queue for asynchronous processing
-    const fetchIntegratorIdSync = events.v3FundsDepositedEvents.length < 1000;
-    if (fetchIntegratorIdSync) {
-      this.appendIntegratorIdToDeposits(events.v3FundsDepositedEvents);
-    }
-
     const storedEvents = await this.storeEvents(events, lastFinalisedBlock);
-
-    if (!fetchIntegratorIdSync) {
-      await this.publishIntegratorIdMessages(storedEvents.deposits);
-    }
-
+    const newInsertedDeposits = indexerDatabaseUtils.filterSaveQueryResults(
+      storedEvents.deposits,
+      SaveQueryResultType.Inserted,
+    );
+    await this.updateNewDepositsWithIntegratorId(newInsertedDeposits);
     await this.spokePoolProcessor.process(storedEvents);
   }
 
@@ -159,7 +164,7 @@ export class SpokePoolIndexerDataHandler implements IndexerDataHandler {
   private async storeEvents(
     params: FetchEventsResult,
     lastFinalisedBlock: number,
-  ) {
+  ): Promise<StoreEventsResult> {
     const { spokePoolClientRepository } = this;
     const {
       v3FundsDepositedEvents,
@@ -223,20 +228,22 @@ export class SpokePoolIndexerDataHandler implements IndexerDataHandler {
     );
   }
 
-  private async appendIntegratorIdToDeposits(
-    deposits: utils.V3FundsDepositedWithIntegradorId[],
+  private async updateNewDepositsWithIntegratorId(
+    deposits: entities.V3FundsDeposited[],
   ) {
-    await across.utils.forEachAsync(
-      deposits,
-      async (deposit, index, deposits) => {
-        const integratorId = await utils.getIntegratorId(
-          this.provider,
-          new Date(deposit.quoteTimestamp * 1000),
-          deposit.transactionHash,
+    await across.utils.forEachAsync(deposits, async (deposit) => {
+      const integratorId = await getIntegratorId(
+        this.provider,
+        deposit.quoteTimestamp,
+        deposit.transactionHash,
+      );
+      if (integratorId) {
+        await this.spokePoolClientRepository.updateDepositEventWithIntegratorId(
+          deposit.id,
+          integratorId,
         );
-        deposits[index] = { ...deposit, integratorId };
-      },
-    );
+      }
+    });
   }
 
   private async publishIntegratorIdMessages(