feat: add optimizations and fix caching

Author: 0xkenj1

packages/data-flow/src/orchestrator.ts

@@ -39,7 +39,7 @@ import { CoreDependencies, DataLoader, delay, IQueue, iStrategyAbi, Queue } from
 
 type TokenWithTimestamps = {
     token: { priceSourceCode: Token["priceSourceCode"] };
-    timestamps: number[];
+    timestamps: TimestampMs[];
 };
 
 /**
@@ -126,13 +126,23 @@ export class Orchestrator {
     }
 
     async run(signal: AbortSignal): Promise<void> {
+        let totalEvents = 0;
+        let processedEvents = 0;
+
         while (!signal.aborted) {
             let event: ProcessorEvent<ContractName, AnyEvent> | undefined;
             try {
                 if (this.eventsQueue.isEmpty()) {
                     const events = await this.getNextEventsBatch();
-                    await this.bulkFetchMetadataAndPricesForBatch(events);
+                    if (
+                        events[0] &&
+                        Math.abs(new Date().getTime() - events[0].blockTimestamp!) >
+                            1000 * 60 * 60 * 0.5 // 30 minutes
+                    ) {
+                        await this.bulkFetchMetadataAndPricesForBatch(events);
+                    }
                     await this.enqueueEvents(events);
+                    totalEvents += events.length;
                 }
 
                 event = this.eventsQueue.pop();
@@ -152,13 +162,19 @@ export class Orchestrator {
                     ...event,
                     rawEvent: event,
                 });
-
+                console.time(`Processing time for event ${event.eventName}`);
                 await this.retryHandler.execute(
                     async () => {
                         await this.handleEvent(event!);
                     },
                     { abortSignal: signal },
                 );
+                console.timeEnd(`Processing time for event ${event.eventName}`);
+                processedEvents++;
+                this.logger.info(`Processed events: ${processedEvents}/${totalEvents}`, {
+                    className: Orchestrator.name,
+                    chainId: this.chainId,
+                });
             } catch (error: unknown) {
                 // TODO: notify
                 if (
@@ -249,6 +265,7 @@ export class Orchestrator {
      */
     private async getNextEventsBatch(): Promise<AnyIndexerFetchedEvent[]> {
         const lastProcessedEvent = await this.eventsRegistry.getLastProcessedEvent(this.chainId);
+
         const blockNumber = lastProcessedEvent?.blockNumber ?? 0;
         const logIndex = lastProcessedEvent?.logIndex ?? 0;
 

packages/data-flow/test/unit/eventsFetcher.spec.ts

@@ -18,7 +18,7 @@ describe("EventsFetcher", () => {
         indexerClientMock = {
             getEventsAfterBlockNumberAndLogIndex: vi.fn(),
             getEvents: vi.fn(),
-            getBlockRangeByChainId: vi.fn(),
+            getBlockRangeTimestampByChainId: vi.fn(),
         };
 
         eventsFetcher = new EventsFetcher(indexerClientMock);

packages/indexer-client/src/interfaces/indexerClient.ts

@@ -34,9 +34,9 @@ export interface IIndexerClient {
     getEvents(params: GetEventsFilters): Promise<AnyIndexerFetchedEvent[]>;
 
     /**
-     * Get the block range by chain id from the indexer service
+     * Get the block range timestamp by chain id from the indexer service
      * @param chainId Id of the chain
      * @returns Block range from the indexer service
      */
-    getBlockRangeByChainId(chainId: ChainId): Promise<{ from: number; to: number }>;
+    getBlockRangeTimestampByChainId(chainId: ChainId): Promise<{ from: number; to: number }>;
 }

packages/indexer-client/src/providers/envioIndexerClient.ts

@@ -267,34 +267,34 @@ export class EnvioIndexerClient implements IIndexerClient {
             methodName,
         });
     }
-    async getBlockRangeByChainId(chainId: ChainId): Promise<{ from: number; to: number }> {
+    async getBlockRangeTimestampByChainId(chainId: ChainId): Promise<{ from: number; to: number }> {
         const response = (await this.client.request(
             gql`
                 query getBlockRangeByChainId($chainId: Int!) {
                     from: raw_events(
                         where: { chain_id: { _eq: $chainId } }
-                        order_by: { block_number: asc }
+                        order_by: { block_timestamp: asc }
                         limit: 1
                     ) {
-                        block_number
+                        block_timestamp
                     }
                     to: raw_events(
                         where: { chain_id: { _eq: $chainId } }
-                        order_by: { block_number: desc }
+                        order_by: { block_timestamp: desc }
                         limit: 1
                     ) {
-                        block_number
+                        block_timestamp
                     }
                 }
             `,
             { chainId },
-        )) as { from: { block_number: number }[]; to: { block_number: number }[] };
+        )) as { from: { block_timestamp: number }[]; to: { block_timestamp: number }[] };
         if (!response.from[0] || !response.to[0]) {
             throw new Error("No block range found");
         }
         return {
-            from: response.from[0].block_number,
-            to: response.to[0].block_number,
+            from: response.from[0].block_timestamp,
+            to: response.to[0].block_timestamp,
         };
     }
 }

packages/pricing/src/constants/index.ts

@@ -1,3 +1,3 @@
 // This is the minimum granularity we can get data with on Enterprise plans of Coingecko
 // Refer to https://support.coingecko.com/hc/en-us/articles/4538747001881-What-granularity-do-you-support-for-historical-data
-export const MIN_GRANULARITY_MS = 300_000; // 5 minutes
+export const MIN_GRANULARITY_MS = 3600_000; // 1 hour

packages/pricing/src/exceptions/noClosePriceFound.exception.ts

@@ -0,0 +1,5 @@
+export class NoClosePriceFound extends Error {
+    constructor() {
+        super(`No close price found`);
+    }
+}