import { Injectable , Logger } from '@nestjs/common' ; import { Cron } from '@nestjs/schedule' ; import { ethers } from 'ethers' ; import { PixelMapEvent } from './entities/pixelMapEvent.entity' ; import { Tile } from './entities/tile.entity' ; import { DataHistory } from './entities/dataHistory.entity' ; import { PurchaseHistory } from './entities/purchaseHistory.entity' ; import { WrappingHistory } from './entities/wrappingHistory.entity' ; import { TransferHistory } from './entities/transferHistory.entity' ; import { DecodedPixelMapTransaction , decodeTransaction , TransactionType } from './utils/decodeTransaction' ; import { initializeEthersJS } from './utils/initializeEthersJS' ; import { exit } from '@nestjs/cli/actions' ; import { EntityRepository , MikroORM , QueryOrder } from '@mikro-orm/core' ; import { InjectRepository , UseRequestContext } from '@mikro-orm/nestjs' ; import { CurrentState , StatesToTrack } from './entities/currentState.entity' ; import { getEvents } from './utils/getEvents' ; import { updateAllTileData } from './utils/updateAllTileData' ; const fs = require ( 'fs' ) ; @ Injectable ( ) export class IngestorService { private readonly logger = new Logger ( IngestorService . name ) ; private currentlyRunningSync = true ; private currentlyIngestingEvents = false ; private updatingCurrentTileData = false ; constructor ( @ InjectRepository ( CurrentState ) private currentState , @ InjectRepository ( PixelMapEvent ) private pixelMapEvent , @ InjectRepository ( Tile ) private tile , @ InjectRepository ( DataHistory ) private dataHistory , @ InjectRepository ( PurchaseHistory ) private purchaseHistory , @ InjectRepository ( WrappingHistory ) private wrappingHistory , @ InjectRepository ( TransferHistory ) private transferHistory , private readonly orm ) { } async onApplicationBootstrap ( ) { this . currentlyIngestingEvents = true ; if ( ! fs . existsSync ( 'cache/metadata' ) ) { fs . mkdirSync ( 'cache/metadata' , { recursive : true } ) ; } if ( ! fs . existsSync ( 'cache/tile' ) ) { fs . mkdirSync ( 'cache/tile' , { recursive : true } ) ; } if ( ( await this . currentState . findOne ( { state : StatesToTrack . INGESTION_LAST_DOWNLOADED_BLOCK } ) ) == undefined ) { this . logger . log ( 'Starting fresh, start of events! (0)' ) ; await this . currentState . persistAndFlush ( [ new CurrentState ( { state : StatesToTrack . INGESTION_LAST_DOWNLOADED_BLOCK , value : 2641527 , } ) , new CurrentState ( { state : StatesToTrack . INGESTION_LAST_PROCESSED_PIXEL_MAP_EVENT , value : 0 , } ) , new CurrentState ( { state : StatesToTrack . NOTIFICATIONS_LAST_PROCESSED_PURCHASE_ID , value : 0 , } ) , new CurrentState ( { state : StatesToTrack . RENDERER_LAST_PROCESSED_DATA_CHANGE , value : 0 , } ) , ] ) ; for ( let i = 0 ; i < 3970 ; i ++ ) { this . logger . log ( 'Initializing tile: ' + i ) ; const tile = await this . tile . create ( { id : i , price : 2 , wrapped : false , image : '' , url : '' , owner : '0x4f4b7e7edf5ec41235624ce207a6ef352aca7050' , } ) ; await this . tile . persistAndFlush ( tile ) ; } await this . tile . flush ( ) ; } this . currentlyIngestingEvents = false ; } /**
     * initializeEthersJS initializes the PixelMap and PixelMapWrapper contracts
     */ @ Cron ( new Date ( Date . now ( ) + 10000 ) ) async initialStartup ( ) { await this . syncBlocks ( ) ; this . currentlyRunningSync = false ; await this . resyncEveryMinute ( ) ; } @ Cron ( '45 * * * * *' ) async resyncEveryMinute ( ) { if ( this . currentlyRunningSync ) { this . logger . log ( 'Already syncing, skipping' ) ; } else { this . logger . debug ( 'Syncing again, one moment.' ) ; this . currentlyRunningSync = true ; await this . syncBlocks ( ) ; this . currentlyRunningSync = false ; } } async syncBlocks ( ) { const { provider } = initializeEthersJS ( ) ; const lastBlock = await this . currentState . findOne ( { state : StatesToTrack . INGESTION_LAST_DOWNLOADED_BLOCK } ) ; let mostRecentBlockNumber = await provider . getBlockNumber ( ) ; mostRecentBlockNumber = mostRecentBlockNumber - 10 ; if ( mostRecentBlockNumber - lastBlock . value > 1000000 ) { mostRecentBlockNumber = lastBlock . value + 1000000 ; } this . logger . log ( 'Downloading raw events from block: ' + lastBlock . value + ' to ' + mostRecentBlockNumber ) ; let rawEvents = [ ] ; try { rawEvents = await getEvents ( lastBlock . value , mostRecentBlockNumber , this . logger ) ; } catch { this . logger . error ( 'Error downloading raw events' ) ; return ; } this . logger . debug ( 'Processing blocks from block: ' + lastBlock . value + ' to ' + mostRecentBlockNumber ) ; await this . processEvents ( rawEvents ) ; lastBlock . value = mostRecentBlockNumber + 1 ; await this . currentState . persistAndFlush ( lastBlock ) ; this . logger . debug ( 'Finished syncing.  Will check again soon.' ) ; } async processEvents ( events ) { const { provider } = initializeEthersJS ( ) ; for ( let i = 0 ; i < events . length ; i ++ ) { const event = events [ i ] ; const transaction = await provider . getTransaction ( event . transactionHash ) ; if ( await this . pixelMapEvent . findOne ( { txHash : event . transactionHash , logIndex : parseInt ( event . logIndex ) } ) ) { this . logger . warn ( 'Already indexed this event, skipping!' ) ; return ; } else { this . logger . log ( 'Saving event at block: ' + transaction . blockNumber ) ; const pixelMapEvent = new PixelMapEvent ( { eventData : event , txHash : event . transactionHash , logIndex : parseInt ( event . logIndex ) , txData : transaction , block : transaction . blockNumber , } ) ; await this . pixelMapEvent . persistAndFlush ( pixelMapEvent ) ; } } await this . pixelMapEvent . flush ( ) ; } @ Cron ( '0 */3 * * *' ) @ UseRequestContext ( ) async updateCurrentTileData ( ) { const { provider , pixelMap , pixelMapWrapper } = initializeEthersJS ( ) ; if ( ! this . updatingCurrentTileData ) { this . updatingCurrentTileData = true ; try { await updateAllTileData ( pixelMap , pixelMapWrapper , this . logger , this . tile ) ; this . updatingCurrentTileData = false ; } catch ( e ) { this . logger . error ( 'Error updating current tile data: ' + e ) ; } } } @ Cron ( '1 * * * * *' ) @ UseRequestContext ( ) async ingestEvents ( ) { if ( ! this . currentlyIngestingEvents ) { const { provider , pixelMap , pixelMapWrapper } = initializeEthersJS ( ) ; this . currentlyIngestingEvents = true ; const lastEvent = await this . currentState . findOne ( { state : StatesToTrack . INGESTION_LAST_PROCESSED_PIXEL_MAP_EVENT , } ) ; const events = await this . pixelMapEvent . find ( { } , { orderBy : { id : QueryOrder . ASC } } ) ; for ( let i = lastEvent . value ; i < events . length ; i ++ ) { const decodedEvent = await decodeTransaction ( pixelMap , pixelMapWrapper , events [ i ] ) ; await this . ingestEvent ( decodedEvent ) ; const percent = Math . round ( ( 100 * ( i + 1 ) ) / events . length ) ; this . logger . verbose ( i + 1 + '/' + events . length + ' events processed (' + percent + '% complete) - ' + events [ i ] . id ) ; lastEvent . value = i ; await this . currentState . persistAndFlush ( lastEvent ) ; } this . currentlyIngestingEvents = false ; } else { this . logger . debug ( 'Already ingesting, not starting again yet' ) ; } } async alreadyIndexed ( repo , txHash , logIndex ) { const existing = await repo . findOne ( { tx : txHash , logIndex : logIndex , } ) ; if ( existing ) { this . logger . warn ( 'Already indexed this update, skipping!' ) ; return true ; } return false ; } async alreadyIndexedWrap ( repo , txHash ) { const existing = await repo . findOne ( { tx : txHash , } ) ; if ( existing ) { this . logger . warn ( 'Already indexed this update, skipping!' ) ; return true ; } return false ; } async ingestEvent ( decodedEvent ) { const tile = await this . tile . findOne ( { id : decodedEvent . location } , [ 'dataHistory' , 'purchaseHistory' , 'wrappingHistory' , 'transferHistory' , ] ) ; switch ( decodedEvent . type ) { case TransactionType . setTile : if ( await this . alreadyIndexed ( this . dataHistory , decodedEvent . txHash , decodedEvent . logIndex ) ) { this . logger . warn ( 'Already indexed this setTile, skipping!' ) ; return ; } this . logger . log ( 'Tile updated at block: ' + decodedEvent . blockNumber + ' (' + decodedEvent . timestamp + ')' ) ; tile . image = decodedEvent . image ? decodedEvent . image . substring ( 0 , 799 ) : tile . image . substring ( 0 , 799 ) ; tile . price = decodedEvent . price ? decodedEvent . price : tile . price ; tile . url = decodedEvent . url ? decodedEvent . url : tile . url ; tile . owner = decodedEvent . from ? decodedEvent . from : tile . owner ; if ( tile . price > 99999 ) { tile . price = 99999 ; } tile . dataHistory . add ( new DataHistory ( { price : tile . price , url : tile . url , tx : decodedEvent . txHash , timeStamp : decodedEvent . timestamp , blockNumber : decodedEvent . blockNumber , image : tile . image . substring ( 0 , 799 ) , updatedBy : decodedEvent . from , logIndex : decodedEvent . logIndex , } ) ) ; await this . tile . persistAndFlush ( tile ) ; break ; case TransactionType . buyTile : if ( await this . alreadyIndexed ( this . purchaseHistory , decodedEvent . txHash , decodedEvent . logIndex ) ) { this . logger . warn ( 'Already indexed this buyTile, skipping!' ) ; return ; } this . logger . log ( 'Tile ' + decodedEvent . location + ' bought at block: ' + decodedEvent . blockNumber + ' (' + decodedEvent . timestamp + ')' ) ; if ( decodedEvent . price <= 0 ) { throw 'Purchase cannot be 0 or less' ; } tile . purchaseHistory . add ( new PurchaseHistory ( { soldBy : tile . owner , purchasedBy : decodedEvent . from , price : decodedEvent . price ? decodedEvent . price : tile . price , tx : decodedEvent . txHash , timeStamp : decodedEvent . timestamp , blockNumber : decodedEvent . blockNumber , logIndex : decodedEvent . logIndex , } ) ) ; tile . owner = decodedEvent . from ; tile . price = 0 ; await this . tile . persistAndFlush ( tile ) ; break ; case TransactionType . wrap : await this . addWrappingHistory ( decodedEvent , tile , true ) ; break ; case TransactionType . unwrap : await this . addWrappingHistory ( decodedEvent , tile , false ) ; break ; case TransactionType . transfer : if ( await this . alreadyIndexed ( this . transferHistory , decodedEvent . txHash , decodedEvent . logIndex ) ) { this . logger . warn ( 'Already indexed this transfer, skipping!' ) ; return ; } this . logger . log ( 'Tile transferred at block: ' + decodedEvent . blockNumber + ' (' + decodedEvent . timestamp + ')' ) ; tile . owner = decodedEvent . from ; tile . transferHistory . add ( new TransferHistory ( { tx : decodedEvent . txHash , timeStamp : decodedEvent . timestamp , blockNumber : decodedEvent . blockNumber , transferredFrom : decodedEvent . to , transferredTo : decodedEvent . from , logIndex : decodedEvent . logIndex , } ) ) ; await this . tile . persistAndFlush ( tile ) ; break ; default : this . logger . error ( 'Failed to account for the following event:' ) ; console . log ( decodedEvent ) ; exit ( ) ; } } async addWrappingHistory ( decodedEvent , tile , wrapped ) { if ( await this . alreadyIndexedWrap ( this . wrappingHistory , decodedEvent . txHash ) ) { const noun = wrapped ? 'wrapping' : 'unwrapping' ; this . logger . warn ( 'Already indexed this ' + noun + ', skipping!' ) ; return ; } this . logger . log ( 'Tile wrapped at block: ' + decodedEvent . blockNumber + ' (' + decodedEvent . timestamp + ')' ) ; tile . wrappingHistory . add ( new WrappingHistory ( { wrapped : wrapped , tx : decodedEvent . txHash , timeStamp : decodedEvent . timestamp , blockNumber : decodedEvent . blockNumber , updatedBy : decodedEvent . from , logIndex : decodedEvent . logIndex , } ) ) ; await this . tile . persistAndFlush ( tile ) ; } }