configurable migrated indices

Author: bogdan-rosianu

config/config.devnet.yaml

@@ -84,6 +84,8 @@ features:
     durationThresholdMs: 5000
     failureCountThreshold: 5
     resetTimeoutMs: 30000
+  elasticMigratedIndices:
+    logs: 'events'
   statusChecker:
     enabled: false
     thresholds:

config/config.e2e.mainnet.yaml

@@ -85,6 +85,8 @@ features:
     durationThresholdMs: 5000
     failureCountThreshold: 5
     resetTimeoutMs: 30000
+  elasticMigratedIndices:
+    logs: 'events'
   statusChecker:
     enabled: false
     thresholds:

config/config.mainnet.yaml

@@ -85,6 +85,8 @@ features:
     durationThresholdMs: 5000
     failureCountThreshold: 5
     resetTimeoutMs: 30000
+  elasticMigratedIndices:
+    logs: 'events'
   statusChecker:
     enabled: false
     thresholds:

config/config.testnet.yaml

@@ -84,6 +84,8 @@ features:
     durationThresholdMs: 5000
     failureCountThreshold: 5
     resetTimeoutMs: 30000
+  elasticMigratedIndices:
+    logs: 'events'
   statusChecker:
     enabled: false
     thresholds:

src/common/api-config/api.config.service.ts

@@ -407,6 +407,10 @@ export class ApiConfigService {
     };
   }
 
+  getElasticMigratedIndicesConfig(): Record<string, string> {
+    return this.configService.get<Record<string, string>>('features.elasticMigratedIndices') ?? {};
+  }
+
   getIsWebsocketApiActive(): boolean {
     return this.configService.get<boolean>('api.websocket') ?? true;
   }

src/common/indexer/elastic/elastic.indexer.service.ts

@@ -591,17 +591,17 @@ export class ElasticIndexerService implements IndexerInterface {
     return query;
   }
 
-  async getTransactionLogs(hashes: string[]): Promise<any[]> {
+  async getTransactionLogs(hashes: string[], eventsIndex: string, txHashField: string): Promise<any[]> {
     const queries = [];
     for (const hash of hashes) {
-      queries.push(QueryType.Match('txHash', hash));
+      queries.push(QueryType.Match(txHashField, hash));
     }
 
     const elasticQueryLogs = ElasticQuery.create()
       .withPagination({ from: 0, size: 10000 })
       .withCondition(QueryConditionOptions.should, queries);
 
-    return await this.elasticService.getList('events', 'id', elasticQueryLogs);
+    return await this.elasticService.getList(eventsIndex, 'id', elasticQueryLogs);
   }
 
   async getTransactionScResults(txHash: string): Promise<any[]> {

src/common/indexer/indexer.interface.ts

@@ -134,7 +134,7 @@ export interface IndexerInterface {
 
   getTokensForAddress(address: string, queryPagination: QueryPagination, filter: TokenFilter): Promise<Token[]>
 
-  getTransactionLogs(hashes: string[]): Promise<ElasticTransactionLogEvent[]>
+  getTransactionLogs(hashes: string[], eventsIndex: string, txHashField: string): Promise<ElasticTransactionLogEvent[]>
 
   getTransactionScResults(txHash: string): Promise<ScResult[]>
 

src/common/indexer/indexer.service.ts

@@ -302,8 +302,8 @@ export class IndexerService implements IndexerInterface {
   }
 
   @LogPerformanceAsync(MetricsEvents.SetIndexerDuration)
-  async getTransactionLogs(hashes: string[]): Promise<ElasticTransactionLogEvent[]> {
-    return await this.indexerInterface.getTransactionLogs(hashes);
+  async getTransactionLogs(hashes: string[], eventsIndex: string, txHashField: string): Promise<ElasticTransactionLogEvent[]> {
+    return await this.indexerInterface.getTransactionLogs(hashes, eventsIndex, txHashField);
   }
 
   @LogPerformanceAsync(MetricsEvents.SetIndexerDuration)

src/endpoints/transactions/transaction.get.service.ts

@@ -20,6 +20,7 @@ import { TransactionOperationType } from "./entities/transaction.operation.type"
 import { QueryPagination } from "src/common/entities/query.pagination";
 import { NftFilter } from "../nfts/entities/nft.filter";
 import { TokenAccount } from "src/common/indexer/entities";
+import { ApiConfigService } from "../../common/api-config/api.config.service";
 
 @Injectable()
 export class TransactionGetService {
@@ -30,6 +31,7 @@ export class TransactionGetService {
     private readonly gatewayService: GatewayService,
     @Inject(forwardRef(() => TokenTransferService))
     private readonly tokenTransferService: TokenTransferService,
+    private readonly apiConfigService: ApiConfigService,
   ) { }
 
   private async tryGetTransactionFromElasticBySenderAndNonce(sender: string, nonce: number): Promise<TransactionDetailed | undefined> {
@@ -51,8 +53,22 @@ export class TransactionGetService {
     return result.map(x => ApiUtils.mergeObjects(new TransactionLog(), x));
   }
 
-  private async getTransactionLogsFromElasticInternal(hashes: string[]): Promise<any[]> {
-    const rawHits = await this.indexerService.getTransactionLogs(hashes);
+  private async getTransactionLogsFromElasticInternal(hashes: string[]) {
+    const esMigratedIndices = this.apiConfigService.getElasticMigratedIndicesConfig();
+    const index = esMigratedIndices?.['logs'] ?? 'logs';
+    if (index === 'events') {
+      return await this.getTransactionLogsFromElasticInternalEventsIndex(hashes);
+    }
+
+    return await this.getTransactionLogsFromElasticInternalLogsIndex(hashes);
+  }
+
+  private async getTransactionLogsFromElasticInternalLogsIndex(hashes: string[]): Promise<any[]> {
+    return await this.indexerService.getTransactionLogs(hashes, 'logs', '_id');
+  }
+
+  private async getTransactionLogsFromElasticInternalEventsIndex(hashes: string[]): Promise<any[]> {
+    const rawHits = await this.indexerService.getTransactionLogs(hashes, 'events', 'txHash');
 
     const logsMap: Map<string, TransactionLog> = new Map();