feat: bulk fetch metadata & prices for events batch

apps/processing/src/services/sharedDependencies.service.ts

@@ -5,6 +5,8 @@ import { CachingPricingProvider, PricingProviderFactory } from "@grants-stack-in
 import {
     createKyselyDatabase,
     IEventRegistryRepository,
+    InMemoryMetadataCache,
+    InMemoryPricingCache,
     IStrategyProcessingCheckpointRepository,
     IStrategyRegistryRepository,
     KyselyApplicationPayoutRepository,
@@ -71,22 +73,33 @@ export class SharedDependenciesService {
             env.DATABASE_SCHEMA,
         );
         const pricingRepository = new KyselyPricingCache(kyselyDatabase, env.DATABASE_SCHEMA);
-        const pricingProvider = PricingProviderFactory.create(env, {
+        const internalPricingProvider = PricingProviderFactory.create(env, {
             logger,
         });
-        const cachedPricingProvider = new CachingPricingProvider(
-            pricingProvider,
+        const dbCachedPricingProvider = new CachingPricingProvider(
+            internalPricingProvider,
             pricingRepository,
             logger,
         );
 
+        const pricingProvider = new CachingPricingProvider(
+            dbCachedPricingProvider,
+            new InMemoryPricingCache(),
+            logger,
+        );
+
         const metadataRepository = new KyselyMetadataCache(kyselyDatabase, env.DATABASE_SCHEMA);
-        const metadataProvider = new IpfsProvider(env.IPFS_GATEWAYS_URL, logger);
-        const cachedMetadataProvider = new CachingMetadataProvider(
-            metadataProvider,
+        const internalMetadataProvider = new IpfsProvider(env.IPFS_GATEWAYS_URL, logger);
+        const dbCachedMetadataProvider = new CachingMetadataProvider(
+            internalMetadataProvider,
             metadataRepository,
             logger,
         );
+        const metadataProvider = new CachingMetadataProvider(
+            dbCachedMetadataProvider,
+            new InMemoryMetadataCache(),
+            logger,
+        );
 
         const eventRegistryRepository = new KyselyEventRegistryRepository(
             kyselyDatabase,
@@ -118,9 +131,9 @@ export class SharedDependenciesService {
                 projectRepository,
                 roundRepository,
                 applicationRepository,
-                pricingProvider: cachedPricingProvider,
+                pricingProvider,
                 donationRepository,
-                metadataProvider: cachedMetadataProvider,
+                metadataProvider,
                 applicationPayoutRepository,
                 transactionManager,
             },

apps/processing/test/unit/sharedDependencies.service.spec.ts

@@ -47,6 +47,8 @@ vi.mock("@grants-stack-indexer/repository", () => ({
     KyselyTransactionManager: vi.fn(),
     KyselyPricingCache: vi.fn(),
     KyselyMetadataCache: vi.fn(),
+    InMemoryPricingCache: vi.fn(),
+    InMemoryMetadataCache: vi.fn(),
 }));
 
 vi.mock("@grants-stack-indexer/pricing", () => ({

packages/data-flow/src/orchestrator.ts

@@ -1,6 +1,7 @@
 import { isNativeError } from "util/types";
 
 import { IIndexerClient } from "@grants-stack-indexer/indexer-client";
+import { TokenPrice } from "@grants-stack-indexer/pricing";
 import {
     existsHandler,
     UnsupportedEventException,
@@ -13,6 +14,7 @@ import {
     AnyIndexerFetchedEvent,
     ChainId,
     ContractName,
+    getToken,
     Hex,
     ILogger,
     isAlloEvent,
@@ -23,6 +25,7 @@ import {
     RetryStrategy,
     StrategyEvent,
     stringify,
+    Token,
 } from "@grants-stack-indexer/shared";
 
 import type { IEventsFetcher, IEventsRegistry, IStrategyRegistry } from "./interfaces/index.js";
@@ -31,6 +34,11 @@ import { EventsProcessor } from "./eventsProcessor.js";
 import { InvalidEvent } from "./exceptions/index.js";
 import { CoreDependencies, DataLoader, delay, IQueue, iStrategyAbi, Queue } from "./internal.js";
 
+type TokenWithTimestamps = {
+    token: Token;
+    timestamps: number[];
+};
+
 /**
  * The Orchestrator is the central coordinator of the data flow system, managing the interaction between
  * three main components:
@@ -116,7 +124,11 @@ export class Orchestrator {
         while (!signal.aborted) {
             let event: ProcessorEvent<ContractName, AnyEvent> | undefined;
             try {
-                if (this.eventsQueue.isEmpty()) await this.enqueueEvents();
+                if (this.eventsQueue.isEmpty()) {
+                    const events = await this.getNextEventsBatch();
+                    await this.bulkFetchMetadataAndPricesForBatch(events);
+                    await this.enqueueEvents(events);
+                }
 
                 event = this.eventsQueue.pop();
 
@@ -197,6 +209,51 @@ export class Orchestrator {
         });
     }
 
+    private async getMetadataFromEvents(events: AnyIndexerFetchedEvent[]): Promise<string[]> {
+        const ids = new Set<string>();
+
+        for (const event of events) {
+            if ("metadata" in event.params) {
+                ids.add(event.params.metadata[1]);
+            }
+        }
+
+        return Array.from(ids);
+    }
+
+    private async getTokensFromEvents(
+        events: AnyIndexerFetchedEvent[],
+    ): Promise<TokenWithTimestamps[]> {
+        const tokenMap = new Map<string, TokenWithTimestamps>();
+
+        for (const event of events) {
+            if (
+                "token" in event.params &&
+                "amount" in event.params &&
+                BigInt(event.params.amount) > 0n
+            ) {
+                const token = getToken(this.chainId, event.params.token);
+                if (!token) continue;
+
+                const existing = tokenMap.get(token.address);
+                if (existing) {
+                    existing.timestamps.push(event.blockTimestamp);
+                } else {
+                    tokenMap.set(token.address, {
+                        token,
+                        timestamps: [event.blockTimestamp],
+                    });
+                }
+            }
+        }
+
+        // Convert timestamps to unique sorted arrays
+        return Array.from(tokenMap.values()).map(({ token, timestamps }) => ({
+            token,
+            timestamps: [...new Set(timestamps)].sort((a, b) => a - b),
+        }));
+    }
+
     /**
      * Sometimes the TimestampsUpdated event is part of the _initialize() function of a strategy.
      * In this case, the event is emitted before the PoolCreated event. We can safely ignore the error
@@ -224,10 +281,7 @@ export class Orchestrator {
         return false;
     }
 
-    /**
-     * Enqueue new events from the events fetcher using the last processed event as a starting point
-     */
-    private async enqueueEvents(): Promise<void> {
+    private async getNextEventsBatch(): Promise<AnyIndexerFetchedEvent[]> {
         const lastProcessedEvent = await this.eventsRegistry.getLastProcessedEvent(this.chainId);
         const blockNumber = lastProcessedEvent?.blockNumber ?? 0;
         const logIndex = lastProcessedEvent?.logIndex ?? 0;
@@ -240,6 +294,33 @@ export class Orchestrator {
             allowPartialLastBlock: false,
         });
 
+        return events;
+    }
+
+    /**
+     * Clear caches and fetch metadata and prices for the batch
+     */
+    private async bulkFetchMetadataAndPricesForBatch(
+        events: AnyIndexerFetchedEvent[],
+    ): Promise<void> {
+        // Clear caches
+        if (this.dependencies.metadataProvider.clearCache) {
+            await this.dependencies.metadataProvider.clearCache();
+        }
+
+        const metadataIds = await this.getMetadataFromEvents(events);
+        const tokens = await this.getTokensFromEvents(events);
+
+        await Promise.allSettled([
+            this.bulkFetchMetadata(metadataIds),
+            this.bulkFetchTokens(tokens),
+        ]);
+    }
+
+    /**
+     * Enqueue events and updates new context for the batch
+     */
+    private async enqueueEvents(events: AnyIndexerFetchedEvent[]): Promise<void> {
         // Clear previous context
         this.eventsByBlockContext.clear();
         for (const event of events) {
@@ -252,6 +333,54 @@ export class Orchestrator {
         this.eventsQueue.push(...events);
     }
 
+    /**
+     * Fetch all possible metadata for the batch
+     */
+    private async bulkFetchMetadata(metadataIds: string[]): Promise<unknown[]> {
+        const results = await Promise.allSettled(
+            metadataIds.map((id) =>
+                this.retryHandler.execute(() =>
+                    this.dependencies.metadataProvider.getMetadata<unknown>(id),
+                ),
+            ),
+        );
+
+        const metadata: unknown[] = [];
+        for (const result of results) {
+            if (result.status === "fulfilled" && result.value) {
+                metadata.push(result.value);
+            }
+        }
+
+        return metadata;
+    }
+
+    /**
+     * Fetch all possible prices for the batch
+     */
+    private async bulkFetchTokens(tokens: TokenWithTimestamps[]): Promise<TokenPrice[]> {
+        const results = await Promise.allSettled(
+            tokens.map(({ token, timestamps }) =>
+                this.retryHandler.execute(async () => {
+                    const prices = await this.dependencies.pricingProvider.getTokenPrices(
+                        token.priceSourceCode,
+                        timestamps,
+                    );
+                    return prices;
+                }),
+            ),
+        );
+
+        const tokenPrices: TokenPrice[] = [];
+        for (const result of results) {
+            if (result.status === "fulfilled" && result.value) {
+                tokenPrices.push(...result.value);
+            }
+        }
+
+        return tokenPrices;
+    }
+
     private async handleEvent(event: ProcessorEvent<ContractName, AnyEvent>): Promise<void> {
         event = await this.enhanceStrategyId(event);
         if (this.isPoolCreated(event)) {

packages/data-flow/src/types/index.ts

@@ -1,3 +1,5 @@
+import { IMetadataProvider } from "@grants-stack-indexer/metadata";
+import { IPricingProvider } from "@grants-stack-indexer/pricing";
 import { ProcessorDependencies } from "@grants-stack-indexer/processors";
 import {
     IApplicationPayoutRepository,
@@ -7,6 +9,7 @@ import {
     IRoundRepository,
     ITransactionManager,
 } from "@grants-stack-indexer/repository";
+import { ICacheable } from "@grants-stack-indexer/shared";
 
 /**
  * The core dependencies for the data flow
@@ -19,6 +22,8 @@ export type CoreDependencies = Pick<
     ProcessorDependencies,
     "evmProvider" | "pricingProvider" | "metadataProvider"
 > & {
+    pricingProvider: IPricingProvider & Partial<ICacheable>;
+    metadataProvider: IMetadataProvider & Partial<ICacheable>;
     roundRepository: IRoundRepository;
     projectRepository: IProjectRepository;
     applicationRepository: IApplicationRepository;