block_processing_env.ts

/* eslint-disable dot-notation */
/*
 * Copyright © 2021 Lisk Foundation
 *
 * See the LICENSE file at the top-level directory of this distribution
 * for licensing information.
 *
 * Unless otherwise agreed in a custom licensing agreement with the Lisk Foundation,
 * no part of this software, including this file, may be copied, modified,
 * propagated, or distributed except according to the terms contained in the
 * LICENSE file.
 *
 * Removal or modification of this copyright notice is prohibited.
 *
 */

import * as fs from 'fs-extra';
import * as path from 'path';
import * as os from 'os';
import {
	Block,
	Chain,
	DataAccess,
	BlockHeader,
	Transaction,
	StateStore,
	standardEventDataSchema,
} from '@liskhq/lisk-chain';
import { Database, StateDB } from '@liskhq/lisk-db';
import { objects } from '@liskhq/lisk-utils';
import { codec, Schema } from '@liskhq/lisk-codec';
import { BaseModule } from '../modules';
import { channelMock } from './mocks';
import {
	defaultConfig,
	getGeneratorPrivateKeyFromDefaultConfig,
	getKeysFromDefaultConfig,
	Keys,
} from './fixtures';
import { removeDB } from './utils';
import { ApplicationConfig, EndpointHandler, EventJSON, GenesisConfig } from '../types';
import { Consensus } from '../engine/consensus';
import { MethodContext, StateMachine } from '../state_machine';
import { Engine } from '../engine';
import {
	createImmutableMethodContext,
	createNewMethodContext,
} from '../state_machine/method_context';
import { blockAssetsJSON } from './fixtures/genesis-asset';
import { ValidatorsModule } from '../modules/validators';
import { TokenModule } from '../modules/token';
import { AuthModule } from '../modules/auth';
import { FeeModule } from '../modules/fee';
import { RandomModule } from '../modules/random';
import { PoSModule } from '../modules/pos';
import { Generator } from '../engine/generator';
import { ABIHandler } from '../abi_handler/abi_handler';
import { generateGenesisBlock } from '../genesis_block';
import { systemDirs } from '../system_dirs';
import { PrefixedStateReadWriter } from '../state_machine/prefixed_state_read_writer';
import { createLogger } from '../logger';
import { MainchainInteroperabilityModule } from '../modules/interoperability';
import { DynamicRewardModule } from '../modules/dynamic_rewards';

type Options = {
	genesis?: GenesisConfig;
	modules?: Record<string, Record<string, unknown>>;
	databasePath?: string;
	passphrase?: string;
};

interface BlockProcessingParams {
	modules?: BaseModule[];
	options?: Options;
	initValidators?: Buffer[];
	logLevel?: string;
}

type DecodedEventJSON = EventJSON & { decodedData: Record<string, unknown> };

export interface BlockProcessingEnv {
	createBlock: (transactions?: Transaction[], timestamp?: number) => Promise<Block>;
	getConsensus: () => Consensus;
	getConsensusStore: () => StateStore;
	getGenerator: () => Generator;
	getGenesisBlock: () => Block;
	getChain: () => Chain;
	getMethodContext: () => MethodContext;
	getBlockchainDB: () => Database;
	process: (block: Block) => Promise<void>;
	processUntilHeight: (height: number) => Promise<void>;
	getLastBlock: () => Block;
	getNextValidatorKeys: (blockHeader: BlockHeader, distance?: number) => Promise<Keys>;
	getDataAccess: () => DataAccess;
	getEvents: (height: number, module?: string, name?: string) => Promise<DecodedEventJSON[]>;
	getChainID: () => Buffer;
	invoke: <T = void>(path: string, params?: Record<string, unknown>) => Promise<T>;
	cleanup: (config: Options) => void;
}

const getAppConfig = (
	genesisConfig?: Partial<GenesisConfig>,
	moduleConfig?: Record<string, Record<string, unknown>>,
): ApplicationConfig => {
	const mergedConfig = objects.mergeDeep(
		{},
		{
			...defaultConfig,
			system: {
				...defaultConfig.system,
				dataPath: path.join(os.tmpdir(), `lisk-framework-test-${Date.now().toString()}`),
			},
			genesis: {
				...defaultConfig.genesis,
				...(genesisConfig ?? {}),
			},
			modules: {
				...(moduleConfig ?? {}),
			},
		},
	) as ApplicationConfig;

	return mergedConfig;
};

const getNextTimestamp = (engine: Engine, previousBlock: BlockHeader, distance = 1) => {
	const previousSlotNumber = engine['_consensus']['_bft'].method.getSlotNumber(
		previousBlock.timestamp,
	);
	return engine['_consensus']['_bft'].method.getSlotTime(previousSlotNumber + distance);
};

const createProcessableBlock = async (
	engine: Engine,
	transactions: Transaction[],
	timestamp?: number,
): Promise<Block> => {
	engine['_logger'].info({ numberOfTransactions: transactions.length }, 'Start generating block');
	// Get previous block and generate valid timestamp, seed reveal, maxHeightPrevoted, reward and maxHeightPreviouslyForged
	const stateStore = new StateStore(engine['_blockchainDB']);
	const previousBlockHeader = engine['_chain'].lastBlock.header;
	const nextTimestamp = timestamp ?? getNextTimestamp(engine, previousBlockHeader);
	const generator = await engine['_consensus']['_bft'].method.getGeneratorAtTimestamp(
		stateStore,
		previousBlockHeader.height + 1,
		nextTimestamp,
	);
	const generatorPrivateKey = getGeneratorPrivateKeyFromDefaultConfig(generator.address);
	for (const tx of transactions) {
		await engine['_generator']['_pool'].add(tx);
	}
	const block = await engine.generateBlock({
		generatorAddress: generator.address,
		height: previousBlockHeader.height + 1,
		privateKey: generatorPrivateKey,
		timestamp: nextTimestamp,
		transactions,
	});
	engine['_logger'].info(
		{ height: block.header.height, numberOfTransactions: block.transactions.length },
		'Generated block',
	);

	return block;
};

export const getBlockProcessingEnv = async (
	params: BlockProcessingParams,
): Promise<BlockProcessingEnv> => {
	const chainID = Buffer.from('00000000', 'hex');
	const appConfig = getAppConfig(
		params.options?.genesis
			? {
					...params.options.genesis,
					chainID: chainID.toString('hex'),
			  }
			: { chainID: chainID.toString('hex') },
		params.options?.modules ?? {},
	);

	const systemDir = systemDirs(appConfig.system.dataPath);

	removeDB(systemDir.data);
	const moduleDB = new Database(path.join(systemDir.data, 'module.db'));
	const stateDB = new StateDB(path.join(systemDir.data, 'state.db'));

	const validatorsModule = new ValidatorsModule();
	const authModule = new AuthModule();
	const tokenModule = new TokenModule();
	const feeModule = new FeeModule();
	const rewardModule = new DynamicRewardModule();
	const randomModule = new RandomModule();
	const posModule = new PoSModule();
	const interopModule = new MainchainInteroperabilityModule();
	const modules = [
		validatorsModule,
		authModule,
		tokenModule,
		feeModule,
		rewardModule,
		randomModule,
		posModule,
	];
	const stateMachine = new StateMachine();
	const logger = createLogger({ name: 'blockProcessingEnv', logLevel: params.logLevel ?? 'none' });

	// resolve dependencies
	tokenModule.addDependencies(interopModule.method, feeModule.method);
	feeModule.addDependencies(tokenModule.method, interopModule.method);
	rewardModule.addDependencies(
		tokenModule.method,
		randomModule.method,
		validatorsModule.method,
		posModule.method,
	);
	posModule.addDependencies(
		randomModule.method,
		validatorsModule.method,
		tokenModule.method,
		feeModule.method,
	);

	// register modules
	stateMachine.registerModule(feeModule);
	stateMachine.registerModule(authModule);
	stateMachine.registerModule(validatorsModule);
	stateMachine.registerModule(tokenModule);
	stateMachine.registerModule(rewardModule);
	stateMachine.registerModule(randomModule);
	stateMachine.registerModule(posModule);
	const blockAssets = blockAssetsJSON.map(asset => ({
		...asset,
		data: codec.fromJSON<Record<string, unknown>>(asset.schema, asset.data),
	}));
	await stateMachine.init(logger, appConfig.genesis, appConfig.modules);
	const genesisBlock = await generateGenesisBlock(stateMachine, logger, {
		timestamp: Math.floor(Date.now() / 1000) - 60 * 60 * 12,
		assets: blockAssets,
		chainID,
	});
	const abiHandler = new ABIHandler({
		channel: channelMock,
		config: appConfig,
		logger,
		stateDB,
		moduleDB,
		modules,
		stateMachine,
		chainID,
	});
	appConfig.genesis.block.blob = genesisBlock.getBytes().toString('hex');
	appConfig.generator.keys.fromFile = path.join(__dirname, './fixtures/keys_fixture.json');
	const engine = new Engine(abiHandler, appConfig);
	await engine['_init']();
	engine['_logger'] = logger;
	engine['_consensus']['_logger'] = logger;

	await abiHandler.init({
		chainID,
		lastBlockHeight: engine['_chain'].lastBlock.header.height,
		lastStateRoot: engine['_chain'].lastBlock.header.stateRoot as Buffer,
	});

	return {
		createBlock: async (transactions: Transaction[] = [], timestamp?: number): Promise<Block> =>
			createProcessableBlock(engine, transactions, timestamp),
		getGenesisBlock: () => genesisBlock,
		getChain: () => engine['_chain'],
		getConsensus: () => engine['_consensus'],
		getConsensusStore: () => new StateStore(engine['_blockchainDB']),
		getGenerator: () => engine['_generator'],
		getMethodContext: () => createNewMethodContext(stateDB.newReadWriter()),
		getBlockchainDB: () => engine['_blockchainDB'],
		process: async (block): Promise<void> => {
			logger.debug(
				{ height: block.header.height, numberOfTransactions: block.transactions.length },
				'Start executing block',
			);
			await engine['_consensus']['_execute'](block, 'peer-id');
			logger.debug(
				{ height: block.header.height, numberOfTransactions: block.transactions.length },
				'Executed block',
			);
		},
		processUntilHeight: async (height): Promise<void> => {
			while (engine['_chain'].lastBlock.header.height < height) {
				const nextBlock = await createProcessableBlock(engine, []);
				await engine['_consensus'].execute(nextBlock);
			}
		},
		getLastBlock: () => engine['_chain'].lastBlock,
		getNextValidatorKeys: async (previousBlockHeader: BlockHeader, distance = 1): Promise<Keys> => {
			const stateStore = new StateStore(engine['_blockchainDB']);
			const nextTimestamp = getNextTimestamp(engine, previousBlockHeader, distance);
			const validator = await engine['_consensus']['_bft'].method.getGeneratorAtTimestamp(
				stateStore,
				previousBlockHeader.height + distance,
				nextTimestamp,
			);
			const keys = getKeysFromDefaultConfig(validator.address);

			return keys;
		},
		getEvents: async (
			height: number,
			module?: string,
			name?: string,
		): Promise<DecodedEventJSON[]> => {
			const handler = engine['_rpcServer']['_getHandler']('chain', 'getEvents');
			if (!handler) {
				throw new Error('invalid');
			}
			const resp = (await handler({
				logger,
				chainID: engine['_chain'].chainID,
				params: { height },
			})) as EventJSON[];
			const eventMetadata = stateMachine['_modules']
				.map(m => ({ module: m.name, ...m.metadata().events } as Record<string, unknown>))
				.flat();
			const results: DecodedEventJSON[] = [];
			for (const event of resp) {
				if (module && event.module !== module) {
					continue;
				}
				if (name && event.name !== name) {
					continue;
				}
				if (event.name === 'commandExecutionResult') {
					results.push({
						...event,
						decodedData: codec.decodeJSON(standardEventDataSchema, Buffer.from(event.data, 'hex')),
					});
					continue;
				}
				const metadata = eventMetadata.find(
					e => e.module === event.module && e.name === event.name,
				);

				results.push({
					...event,
					decodedData: metadata
						? codec.decodeJSON(metadata.schema as Schema, Buffer.from(event.data, 'hex'))
						: {},
				});
			}
			return results;
		},
		async invoke<T = void>(func: string, input: Record<string, unknown> = {}): Promise<T> {
			const [namespace, method] = func.split('_');
			const handler = engine['_rpcServer']['_getHandler'](namespace, method);
			if (handler) {
				const resp = (await handler({
					logger,
					chainID: engine['_chain'].chainID,
					params: input,
				})) as T;
				return resp;
			}
			const moduleIndex = modules.findIndex(mod => mod.name === namespace);
			if (moduleIndex < 0) {
				throw new Error(`namespace ${namespace} is not registered`);
			}
			const moduleHandler = modules[moduleIndex].endpoint[method] as EndpointHandler | undefined;
			if (!moduleHandler) {
				throw new Error(`Method ${method} in namespace ${namespace} is not registered`);
			}
			const bindedHandler = moduleHandler.bind(modules[moduleIndex].endpoint);

			const stateStore = new PrefixedStateReadWriter(stateDB.newReadWriter());
			try {
				const result = await bindedHandler({
					getStore: (moduleID: Buffer, storePrefix: Buffer) =>
						stateStore.getStore(moduleID, storePrefix),
					getImmutableMethodContext: () => createImmutableMethodContext(stateStore),
					logger: engine['_logger'],
					chainID: engine['_chain'].chainID,
					params: input,
				});

				return result as T;
			} finally {
				stateStore.inner.close();
			}
		},
		getChainID: () => chainID,
		getDataAccess: () => engine['_chain'].dataAccess,
		cleanup: (_val): void => {
			engine['_closeDB']();
			moduleDB.close();
			stateDB.close();
			fs.removeSync(systemDir.data);
		},
	};
};