validator-logging/server/StatisticsCollector.ts

import {ApiPromise, WsProvider} from "@polkadot/api";
import {types} from '@joystream/types'
import { HeaderExtended } from '@polkadot/api-derive/type/types';
import {
    AccountId,
    EraIndex,
    EventRecord,
    Hash,
    Moment
} from "@polkadot/types/interfaces";

import {
    CacheEvent,
    Statistics,
} from "./types";

import {Option, u32, Vec} from "@polkadot/types";
import {RewardRelationship, RewardRelationshipId} from "@joystream/types/recurring-rewards";

const fsSync = require('fs');
const fs = fsSync.promises;
const PROVIDER_URL = process.env.PROVIDER_URL; //"wss://rome-rpc-endpoint.joystream.org:9944";

const CACHE_FOLDER = "cache";

export class StatisticsCollector {

    private api?: ApiPromise;
    private blocksEventsCache: Map<number, CacheEvent[]>;
    private statistics: Statistics;

    constructor() {
        this.blocksEventsCache = new Map<number, CacheEvent[]>();
        this.statistics = new Statistics();
    }

    async getStatistics(startBlock: number, endBlock: number): Promise<Statistics> {
        this.api = await StatisticsCollector.connectApi();

        let startHash = (await this.api.rpc.chain.getBlockHash(startBlock)) as Hash;
        let endHash = (await this.api.rpc.chain.getBlockHash(endBlock)) as Hash;

        this.statistics.startBlock = startBlock;
        this.statistics.endBlock = endBlock;
        this.statistics.newBlocks = endBlock - startBlock;
        await this.buildBlocksEventCache(startBlock, endBlock);
        await this.fillBasicInfo(startHash, endHash);
        console.log("1");
        await this.fillValidatorInfo(startHash, endHash);
        console.log("2");
        
        await this.api.disconnect();
        return this.statistics;
    }

    async fillBasicInfo(startHash: Hash, endHash: Hash) {
        let startDate = (await this.api.query.timestamp.now.at(startHash)) as Moment;
        let endDate = (await this.api.query.timestamp.now.at(endHash)) as Moment;
        this.statistics.dateStart = new Date(startDate.toNumber()).toLocaleDateString("en-US");
        this.statistics.dateEnd = new Date(endDate.toNumber()).toLocaleDateString("en-US");
    }

    async computeReward(roundNrBlocks: number, rewardRelationshipIds: RewardRelationshipId[], hash: Hash) {
        let recurringRewards = await Promise.all(rewardRelationshipIds.map(async (rewardRelationshipId) => {
            return await this.api.query.recurringRewards.rewardRelationships.at(hash, rewardRelationshipId) as RewardRelationship;
        }));

        let rewardPerBlock = 0;
        for (let recurringReward of recurringRewards) {
            const amount = recurringReward.amount_per_payout.toNumber();
            const payoutInterval = recurringReward.payout_interval.unwrapOr(null);

            if (amount && payoutInterval) {
                rewardPerBlock += amount / payoutInterval;
            }

        }
        return rewardPerBlock * roundNrBlocks;
    }




    async fillValidatorInfo(startHash: Hash, endHash: Hash) {
        let startTimestamp = await this.api.query.timestamp.now.at(startHash) as unknown as Moment;
        let endTimestamp = await this.api.query.timestamp.now.at(endHash) as unknown as Moment;
        let avgBlockProduction = (((endTimestamp.toNumber() - startTimestamp.toNumber())
            / 1000) / this.statistics.newBlocks);
        this.statistics.avgBlockProduction = Number(avgBlockProduction.toFixed(2));

        let maxStartValidators = (await this.api.query.staking.validatorCount.at(startHash) as u32).toNumber();
        let startValidators = await this.findActiveValidators(startHash, false);
        this.statistics.startValidators = startValidators.length + " / " + maxStartValidators;

        let maxEndValidators = (await this.api.query.staking.validatorCount.at(endHash) as u32).toNumber();
        let endValidators = await this.findActiveValidators(endHash, true);
        this.statistics.endValidators = endValidators.length + " / " + maxEndValidators;

        let aaa = (await this.api.query.staking.erasStakers.at(startHash) as u32).toNumber();
        let bbb = (await this.api.query.staking.erasStakers.at(endHash) as u32).toNumber();


        const startEra = await this.api.query.staking.currentEra.at(startHash) as Option<EraIndex>;
        this.statistics.startValidatorsStake = (await this.api.query.staking.erasTotalStake.at(startHash, startEra.unwrap())).toNumber();

        const endEra = await this.api.query.staking.currentEra.at(endHash) as Option<EraIndex>;
        this.statistics.endValidatorsStake = (await this.api.query.staking.erasTotalStake.at(endHash, endEra.unwrap())).toNumber();

    }

    async findActiveValidators(hash: Hash, searchPreviousBlocks: boolean): Promise<AccountId[]> {
        const block = await this.api.rpc.chain.getBlock(hash);

        let currentBlockNr = block.block.header.number.toNumber();
        let activeValidators;
        do {
            let currentHash = (await this.api.rpc.chain.getBlockHash(currentBlockNr)) as Hash;
            let allValidators = await this.api.query.staking.snapshotValidators.at(currentHash) as Option<Vec<AccountId>>;
            if (!allValidators.isEmpty) {
                let max = (await this.api.query.staking.validatorCount.at(currentHash)).toNumber();
                activeValidators = Array.from(allValidators.unwrap()).slice(0, max);
            }

            if (searchPreviousBlocks) {
                --currentBlockNr;
            } else {
                ++currentBlockNr;
            }

        } while (activeValidators == undefined);
        return activeValidators;
    }

    async buildBlocksEventCache(startBlock: number, endBlock: number) {
        let cacheFile = CACHE_FOLDER + '/' + startBlock + '-' + endBlock + '.json';
        let exists = await fs.access(cacheFile, fsSync.constants.R_OK).then(() => true)
            .catch(() => false);
        // let exists = false;
        if (!exists) {
            console.log('Building events cache...');
            for (let i = startBlock; i < endBlock; ++i) {
                process.stdout.write('\rCaching block: ' + i + ' until ' + endBlock);
                const blockHash: Hash = await this.api.rpc.chain.getBlockHash(i);
                // const signedBlock = await this.api.rpc.chain.getBlock(blockHash);
                const era = await this.api.query.staking.activeEra.at(blockHash)
                console.log(`Era=${era}`);
                const extendedHeader = await this.api.derive.chain.getHeader(blockHash) as HeaderExtended;
                console.log(`#${extendedHeader.number}: ${extendedHeader.author}`);
                let eventRecord = await this.api.query.system.events.at(blockHash) as Vec<EventRecord>;
                let cacheEvents = new Array<CacheEvent>();
                for (let event of eventRecord) {
                    cacheEvents.push(new CacheEvent(event.event.section, event.event.method, event.event.data));
                }
                this.blocksEventsCache.set(i, cacheEvents);
            }

            console.log('\nFinish events cache...');
            await fs.writeFile(cacheFile, JSON.stringify(Array.from(this.blocksEventsCache.entries()), null, 2));
        } else {
            console.log('Cache file found, loading it...');
            let fileData = await fs.readFile(cacheFile);
            this.blocksEventsCache = new Map(JSON.parse(fileData));
            console.log('Cache file loaded...');
        }
    }

    static async connectApi(): Promise<ApiPromise> {
        // const provider = new WsProvider('wss://testnet.joystream.org:9944');
        const provider = new WsProvider(PROVIDER_URL);

        // Create the API and wait until ready
        return await ApiPromise.create({provider, types});
    }
}