Merge branch 'refactor_indexer_logic'

# Conflicts:
#	src/app.controller.ts

Author: ArtemKolodko

src/app.controller.ts

@@ -6,8 +6,14 @@ import {
   Logger,
   NotFoundException,
   Post,
-  Query, UploadedFile, UseInterceptors,
-  Headers, UseGuards, Request, ParseFilePipe, MaxFileSizeValidator, FileTypeValidator
+  Query,
+  UploadedFile,
+  UseInterceptors,
+  UseGuards,
+  Request,
+  ParseFilePipe,
+  MaxFileSizeValidator,
+  FileTypeValidator
 } from '@nestjs/common';
 import {ApiBearerAuth, ApiTags} from '@nestjs/swagger';
 import { ConfigService } from '@nestjs/config';
@@ -26,6 +32,7 @@ import {plainToInstance} from "class-transformer";
 import {JwtUserAccount} from "./entities/user-account.entity";
 import {GetWinnerLiquidityProvisionsDto} from "./dto/winner.liquidity.dto";
 import {CacheTTL} from "@nestjs/common/cache";
+import {IndexerService} from "./indexer/indexer.service";
 import {GetCompetitionsDto} from "./dto/competition.dto";
 
 @SkipThrottle()
@@ -37,16 +44,20 @@ export class AppController {
     private readonly configService: ConfigService,
     private readonly appService: AppService,
     private readonly userService: UserService,
-    private readonly gCloudService: GcloudService
+    private readonly gCloudService: GcloudService,
+    private readonly indexerService: IndexerService,
   ) {}
   @Get('/version')
   getVersion() {
     return this.configService.get('version');
   }
 
   @Get('/status')
-  getStatus() {
-    return 'OK';
+  async getStatus() {
+    const latestIndexedBlock = await this.indexerService.getLatestIndexedBlockNumber();
+    return {
+      latestIndexedBlock,
+    }
   }
 
   @CacheTTL(200)
@@ -130,7 +141,6 @@ export class AppController {
         ]
       })
     ) file: Express.Multer.File,
-    @Headers() headers
   ) {
     if(!req.user) {
       throw new BadRequestException('InvalidJWT')

src/config/index.ts

@@ -5,22 +5,6 @@ import {existsSync, readFileSync} from "fs";
 const parseStringArray = (value: string) =>
   value.split(',').map(item => item.trim().toLowerCase()).filter(_ => _)
 
-const readTestPrivateKey = () => {
-  const filePath = pathResolve(__dirname, '..', '../keypair/private.pem')
-  if(existsSync(filePath)) {
-    return readFileSync(pathResolve(__dirname, '..', '../keypair/private.pem'))
-  }
-  return ''
-}
-
-const readTestPublicKey = () => {
-  const filePath = pathResolve(__dirname, '..', '../keypair/public.pem')
-  if(existsSync(filePath)) {
-    return readFileSync(filePath)
-  }
-  return ''
-}
-
 const getGoogleCloudConfig = () => {
   return {
     "type": "service_account",

src/entities/token.entity.ts

@@ -58,10 +58,6 @@ export class Token {
   @Column({ type: 'json', nullable: true })
   uriData: TokenMetadata | null;
 
-  // @ApiProperty()
-  // @Column({ type: 'integer' })
-  // competitionId: number;
-
   @ManyToOne(() => CompetitionEntity, {
     eager: true
   })

src/gcloud/gcloud.service.ts

@@ -1,18 +1,18 @@
-import {Injectable, Logger} from '@nestjs/common';
+import {Injectable} from '@nestjs/common';
 import {ConfigService} from "@nestjs/config";
 import {Storage} from "@google-cloud/storage";
 import {AddTokenMetadataDto} from "../dto/metadata.dto";
+import {JWTInput} from "google-auth-library/build/src/auth/credentials";
 
 @Injectable()
 export class GcloudService {
-  private readonly logger = new Logger(GcloudService.name);
   private readonly storage: Storage;
 
   constructor(private readonly configService: ConfigService) {
+    const googleCloudConfig = this.configService.get<JWTInput>('GOOGLE_CLOUD_CONFIG')
     this.storage = new Storage({
-      projectId: 'pumpfun-440412',
-      // keyFilename: serviceKey,
-      credentials: this.configService.get('GOOGLE_CLOUD_CONFIG')
+      projectId: googleCloudConfig.project_id,
+      credentials: googleCloudConfig
     })
   }
 

src/indexer/indexer.service.ts

@@ -1,6 +1,6 @@
 import {Injectable, Logger} from '@nestjs/common';
 import {Contract, ContractAbi, EventLog, Web3} from "web3";
-import {TokenMetadata, TradeType} from "../types";
+import {ProtocolEvent, TokenMetadata, TradeType} from "../types";
 import axios from "axios";
 import * as process from "process";
 import {
@@ -27,7 +27,8 @@ export class IndexerService {
   private readonly web3: Web3
   private readonly accountAddress: string
   private readonly tokenFactoryContract: Contract<ContractAbi>
-  private readonly blocksIndexingRange = 1000
+  private readonly maxBlocksRange = 1000
+  private readonly maxBlocksBatchSize = 5
 
   constructor(
     private configService: ConfigService,
@@ -37,32 +38,30 @@ export class IndexerService {
   ) {
     const rpcUrl = configService.get('RPC_URL')
     const contractAddress = configService.get('TOKEN_FACTORY_ADDRESS')
-    const initialBlockNumber = configService.get('INDEXER_INITIAL_BLOCK_NUMBER')
+    const privateKey = configService.get('SERVICE_PRIVATE_KEY')
 
     if(!contractAddress) {
       this.logger.error(`[TOKEN_FACTORY_ADDRESS] is missing but required, exit`)
       process.exit(1)
     }
 
-    if(!initialBlockNumber) {
-      this.logger.error(`[INDEXER_INITIAL_BLOCK_NUMBER] is missing but required, exit`)
+    if(!privateKey) {
+      this.logger.error(`[SERVICE_PRIVATE_KEY] is missing but required, exit`)
       process.exit(1)
     }
 
     this.logger.log(`Starting app service, RPC_URL=${
       rpcUrl
     }, TOKEN_FACTORY_ADDRESS=${
       contractAddress
-    }, INDEXER_INITIAL_BLOCK_NUMBER=${
-      initialBlockNumber
     }`)
 
     this.web3 = new Web3(rpcUrl);
-    const account = this.web3.eth.accounts.privateKeyToAccount(configService.get('SERVICE_PRIVATE_KEY'))
+    const account = this.web3.eth.accounts.privateKeyToAccount(privateKey)
     this.accountAddress = account.address
     this.web3.eth.accounts.wallet.add(account);
-    this.logger.log(`Service account address=${account.address}`)
     this.tokenFactoryContract = new this.web3.eth.Contract(TokenFactoryABI, contractAddress);
+    this.logger.log(`Service account address=${account.address}`)
     this.bootstrap().then(
       () => {
         this.eventsTrackingLoop()
@@ -410,7 +409,7 @@ export class IndexerService {
     this.logger.log(`NewCompetitionStarted: competitionId=${competitionId}, timestamp=${timestamp}, txnHash=${txnHash}`);
   }
 
-  private async getLatestIndexedBlockNumber() {
+  public async getLatestIndexedBlockNumber() {
     const indexerState = await this.dataSource.manager.findOne(IndexerState, {
       where: {},
     })
@@ -478,23 +477,14 @@ export class IndexerService {
       })
     ]) as EventLog[][]
 
-    // concat and sort all events by block number and transaction index
-    const protocolEvents: { data: EventLog; type: string }[] = tokenCreatedEvents
+    return tokenCreatedEvents
       .map(data => ({ type: 'create_token', data }))
       .concat(...buyEvents.map(data => ({ type: 'buy', data })))
       .concat(...sellEvents.map(data => ({ type: 'sell', data })))
       .concat(...setWinnerEvents.map(data => ({ type: 'set_winner', data })))
       .concat(...burnAndSetWinnerEvents.map(data => ({ type: 'burn_token_and_set_winner', data })))
       .concat(...winnerLiquidityEvents.map(data => ({ type: 'winner_liquidity', data })))
       .concat(...newCompetitionEvents.map(data => ({ type: 'new_competition', data })))
-      .sort((a, b) => {
-        const blockNumberDiff = Number(a.data.blockNumber) - Number(b.data.blockNumber)
-        if(blockNumberDiff !== 0) {
-          return blockNumberDiff
-        }
-        return Number(a.data.transactionIndex) - Number(b.data.transactionIndex)
-      })
-    return protocolEvents
   }
 
   async eventsTrackingLoop() {
@@ -506,13 +496,34 @@ export class IndexerService {
 
     try {
       const blockchainBlockNumber = +(String(await this.web3.eth.getBlockNumber()))
-      toBlock = fromBlock + this.blocksIndexingRange - 1
+      toBlock = fromBlock + this.maxBlocksRange * this.maxBlocksBatchSize - 1
       if(toBlock > blockchainBlockNumber) {
         toBlock = blockchainBlockNumber
       }
 
       if(toBlock - fromBlock >= 1) {
-        const protocolEvents = await this.getEventsFromBlocksRange(fromBlock, toBlock)
+        const delta = toBlock - fromBlock
+        const numberOfBatches = Math.ceil(delta / this.maxBlocksRange)
+
+        const protocolEventsBatch = await Promise.all(
+          new Array(numberOfBatches)
+            .fill(null)
+            .map(async (_, index, arr) => {
+              const batchFromBlock = fromBlock + index * this.maxBlocksRange
+              const batchToBlock = Math.min(batchFromBlock + this.maxBlocksRange, toBlock)
+              return await this.getEventsFromBlocksRange(batchFromBlock, batchToBlock)
+            })
+        )
+
+        const protocolEvents = protocolEventsBatch
+          .flat()
+          .sort((a, b) => {
+            const blockNumberDiff = Number(a.data.blockNumber) - Number(b.data.blockNumber)
+            if(blockNumberDiff !== 0) {
+              return blockNumberDiff
+            }
+            return Number(a.data.transactionIndex) - Number(b.data.transactionIndex)
+          })
 
         await this.dataSource.manager.transaction(async (transactionalEntityManager) => {
           for(const protocolEvent of protocolEvents) {

src/types.ts

@@ -27,3 +27,10 @@ export interface Candle {
   volume: string
   time: string
 }
+
+export interface ProtocolEvent {
+  data: EventLog
+  type: ProtocolEventType
+}
+
+export type ProtocolEventType = 'create_token' | 'buy' | 'sell' | 'set_winner' | 'burn_token_and_set_winner' | 'winner_liquidity' | 'new_competition'