Merge pull request #140 from lidofinance/develop

Develop

package.json

@@ -47,6 +47,7 @@
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.0",
     "ethers": "^5.4.7",
+    "glob": "^7.1.2",
     "kafkajs": "^1.15.0",
     "nest-winston": "^1.6.1",
     "node-abort-controller": "^3.0.1",
@@ -106,4 +107,4 @@
     "coverageDirectory": "../coverage",
     "testEnvironment": "node"
   }
-}
+}

src/cache/cache.constants.ts

@@ -2,3 +2,4 @@ export const CACHE_DIR = 'cache';
 
 export const CACHE_FILE_NAME = 'cacheFileName';
 export const CACHE_DEFAULT_VALUE = 'cacheDefaultValue';
+export const CACHE_BATCH_SIZE = 'cacheBatchSize';

src/cache/cache.module.ts

@@ -1,12 +1,16 @@
 import { DynamicModule, Module } from '@nestjs/common';
-import { CACHE_DEFAULT_VALUE, CACHE_FILE_NAME } from 'cache';
+import { CACHE_BATCH_SIZE, CACHE_DEFAULT_VALUE, CACHE_FILE_NAME } from 'cache';
 import { ProviderModule } from 'provider';
 import { CACHE_DIR } from './cache.constants';
 import { CacheService } from './cache.service';
 
 @Module({})
 export class CacheModule {
-  static register(fileName: string, defaultValue: unknown): DynamicModule {
+  static register(
+    fileName: string,
+    batchSize: number,
+    defaultValue: unknown,
+  ): DynamicModule {
     return {
       module: CacheModule,
       imports: [ProviderModule],
@@ -20,6 +24,10 @@ export class CacheModule {
           provide: CACHE_FILE_NAME,
           useValue: fileName,
         },
+        {
+          provide: CACHE_BATCH_SIZE,
+          useValue: batchSize,
+        },
         {
           provide: CACHE_DEFAULT_VALUE,
           useValue: defaultValue,

src/cache/cache.service.spec.ts

@@ -6,16 +6,24 @@ import { CacheModule } from 'cache';
 import { CacheService } from './cache.service';
 
 describe('CacheService', () => {
-  const defaultCacheValue = {};
+  const defaultCacheValue = {
+    headers: {},
+    data: [] as any[],
+  };
+
+  const batchSize = 10;
+
+  type C = typeof defaultCacheValue;
+
   const cacheFile = 'test.json';
-  let cacheService: CacheService<typeof defaultCacheValue>;
+  let cacheService: CacheService<C['headers'], C['data'], C>;
 
   beforeEach(async () => {
     const moduleRef = await Test.createTestingModule({
       imports: [
         ConfigModule.forRoot(),
         MockProviderModule.forRoot(),
-        CacheModule.register(cacheFile, defaultCacheValue),
+        CacheModule.register(cacheFile, batchSize, defaultCacheValue),
         LoggerModule,
       ],
     }).compile();
@@ -32,11 +40,11 @@ describe('CacheService', () => {
   describe('getCache, setCache', () => {
     it('should return default cache', async () => {
       const result = await cacheService.getCache();
-      expect(result).toBe(defaultCacheValue);
+      expect(result).toEqual(defaultCacheValue);
     });
 
     it('should return saved cache', async () => {
-      const expected = { foo: 'bar' };
+      const expected = { headers: {}, data: [{ foo: 'bar' }] };
 
       await cacheService.setCache(expected);
       const result = await cacheService.getCache();

src/cache/cache.service.ts

@@ -1,40 +1,50 @@
 import { Inject, Injectable } from '@nestjs/common';
 import { readFile, writeFile, unlink, mkdir } from 'fs/promises';
 import { join } from 'path';
+import { promisify } from 'util';
+import * as gl from 'glob';
 import {
   CACHE_DIR,
   CACHE_DEFAULT_VALUE,
   CACHE_FILE_NAME,
+  CACHE_BATCH_SIZE,
 } from './cache.constants';
 import { ProviderService } from 'provider';
 
+const glob = promisify(gl.glob);
+
 @Injectable()
-export class CacheService<T extends unknown> {
+export class CacheService<
+  H extends unknown,
+  D extends unknown,
+  T extends { headers: H; data: D[] } = { headers: H; data: D[] },
+> {
   constructor(
     private providerService: ProviderService,
     @Inject(CACHE_DIR) private cacheDir: string,
     @Inject(CACHE_FILE_NAME) private cacheFile: string,
+    @Inject(CACHE_BATCH_SIZE) private cacheBatchSize: number,
     @Inject(CACHE_DEFAULT_VALUE) private cacheDefaultValue: T,
   ) {}
 
   private cache: T | null = null;
 
   public async getCache(): Promise<T> {
     if (!this.cache) {
-      this.cache = await this.getCacheFromFile();
+      this.cache = await this.getCacheFromFiles();
     }
 
     return this.cache;
   }
 
   public async setCache(cache: T): Promise<void> {
     this.cache = cache;
-    return await this.saveCacheToFile();
+    return await this.saveCacheToFiles();
   }
 
   public async deleteCache(): Promise<void> {
     this.cache = null;
-    return await this.deleteCacheFile();
+    return await this.deleteCacheFiles();
   }
 
   private async getCacheDirPath(): Promise<string> {
@@ -44,34 +54,75 @@ export class CacheService<T extends unknown> {
     return join(this.cacheDir, networkDir);
   }
 
-  private async getCacheFilePath(): Promise<string> {
-    const dir = await this.getCacheDirPath();
-    return join(dir, this.cacheFile);
+  private getCacheFileName(batchIndex: number): string {
+    return `${batchIndex}.${this.cacheFile}`;
+  }
+
+  private async getCacheFilePaths(): Promise<string[]> {
+    const dirPath = await this.getCacheDirPath();
+    const result = await glob(`*([0-9]).${this.cacheFile}`, { cwd: dirPath });
+
+    return result
+      .sort((a, b) => parseInt(a) - parseInt(b))
+      .map((filePath) => join(dirPath, filePath));
   }
 
-  private async getCacheFromFile(): Promise<T> {
+  private async getCacheFromFiles(): Promise<T> {
     try {
-      const filePath = await this.getCacheFilePath();
-      const content = await readFile(filePath);
-      return JSON.parse(String(content));
+      const filePaths = await this.getCacheFilePaths();
+
+      let headers = this.cacheDefaultValue.headers as H;
+      let data = [] as D[];
+
+      await Promise.all(
+        filePaths.map(async (filePath) => {
+          const content = await readFile(filePath);
+          const parsed = JSON.parse(String(content));
+
+          if (
+            JSON.stringify(headers) !== JSON.stringify(parsed.headers) &&
+            headers !== this.cacheDefaultValue.headers
+          ) {
+            throw new Error('Headers are not equal');
+          }
+
+          headers = parsed.headers;
+          data = data.concat(parsed.data);
+        }),
+      );
+
+      return { headers, data } as T;
     } catch (error) {
       return this.cacheDefaultValue;
     }
   }
 
-  private async saveCacheToFile(): Promise<void> {
+  private async saveCacheToFiles(): Promise<void> {
+    if (!this.cache) throw new Error('Cache is not set');
+
+    const { headers, data } = this.cache;
+
     const dirPath = await this.getCacheDirPath();
-    const filePath = await this.getCacheFilePath();
     await mkdir(dirPath, { recursive: true });
 
-    return await writeFile(filePath, JSON.stringify(this.cache));
+    await this.deleteCacheFiles();
+
+    const totalBatches = Math.ceil(data.length / this.cacheBatchSize);
+
+    for (let batchIndex = 0; batchIndex < totalBatches; batchIndex++) {
+      const from = batchIndex * this.cacheBatchSize;
+      const to = (batchIndex + 1) * this.cacheBatchSize;
+      const batchedData = data.slice(from, to);
+
+      const filePath = join(dirPath, this.getCacheFileName(batchIndex));
+      await writeFile(filePath, JSON.stringify({ headers, data: batchedData }));
+    }
   }
 
-  private async deleteCacheFile(): Promise<void> {
+  private async deleteCacheFiles(): Promise<void> {
     try {
-      const filePath = await this.getCacheFilePath();
-
-      return await unlink(filePath);
+      const filePaths = await this.getCacheFilePaths();
+      await Promise.all(filePaths.map(async (filePath) => unlink(filePath)));
     } catch (error) {}
   }
 }

src/contracts/deposit/deposit.constants.ts

@@ -1,5 +1,4 @@
 import { CHAINS } from '@lido-sdk/constants';
-import { DepositEventGroup } from './interfaces';
 
 export const DEPLOYMENT_BLOCK_NETWORK: {
   [key in CHAINS]?: number;
@@ -21,10 +20,13 @@ export const DEPOSIT_EVENTS_STEP = 10_000;
 export const DEPOSIT_EVENTS_CACHE_UPDATE_BLOCK_RATE = 10;
 
 export const DEPOSIT_CACHE_FILE_NAME = 'deposit.events.json';
+export const DEPOSIT_CACHE_BATCH_SIZE = 100_000;
 
-export const DEPOSIT_CACHE_DEFAULT: DepositEventGroup = Object.freeze({
-  version: '-1',
-  startBlock: 0,
-  endBlock: 0,
-  events: [],
+export const DEPOSIT_CACHE_DEFAULT = Object.freeze({
+  headers: {
+    version: '-1',
+    startBlock: 0,
+    endBlock: 0,
+  },
+  data: [],
 });

src/contracts/deposit/deposit.module.ts

@@ -4,6 +4,7 @@ import { CacheModule } from 'cache';
 import { BlsModule } from 'bls';
 import { DepositService } from './deposit.service';
 import {
+  DEPOSIT_CACHE_BATCH_SIZE,
   DEPOSIT_CACHE_DEFAULT,
   DEPOSIT_CACHE_FILE_NAME,
 } from './deposit.constants';
@@ -12,7 +13,11 @@ import {
   imports: [
     BlsModule,
     SecurityModule,
-    CacheModule.register(DEPOSIT_CACHE_FILE_NAME, DEPOSIT_CACHE_DEFAULT),
+    CacheModule.register(
+      DEPOSIT_CACHE_FILE_NAME,
+      DEPOSIT_CACHE_BATCH_SIZE,
+      DEPOSIT_CACHE_DEFAULT,
+    ),
   ],
   providers: [DepositService],
   exports: [DepositService],