Lisk SDK v6を使ったブロックチェーンアプリの作り方　その６

はじめに

こんにちは万博おじです。
今回はイベントとプラグインの作成編です。
この回で触れたファイルについては全内容を記事の最後に記載しておきますのでわからなくなったら参考にどうぞ。

前回の記事はこちら

イベントの作成

参考公式ドキュメント：5. How to create a blockchain event

1. イベントの作成

src/app/modules/hello/events/ 直下に new_hello.tsを作成し以下のように編集します。

import { BaseEvent } from 'lisk-sdk';

export interface NewHelloEventData {
	senderAddress: Buffer;
	message: string;
}

export const newHelloEventSchema = {
	$id: '/hello/events/new_hello',
	type: 'object',
	required: ['senderAddress', 'message'],
	properties: {
		senderAddress: {
			dataType: 'bytes',
			fieldNumber: 1,
		},
		message: {
			dataType: 'string',
			fieldNumber: 2,
		},
	},
};

export class NewHelloEvent extends BaseEvent<NewHelloEventData> {
	public schema = newHelloEventSchema;
}

イベントはブロックチェーンで発生したイベントを300ブロックの間オンチェーンに保持する仕組みです。
後述のプラグインの作成で使用するので作成するのを忘れないようにしましょう。

なお、このイベントの内容は「createHello」イベントの送信者のアドレス(senderAddress)およびコマンド実行の際に設定したメッセージ(message)を保持するものです。
スキーマを用意して、それにあわせてインターフェースを作成するのはここまでやっている方には見慣れたものですね！

「createHello」コマンドについては以下の記事を参照

2. イベントの登録

イベントを利用するためにsrc/app/modules/hello/module.tsを以下のように修正します。

～省略～
import { NewHelloEvent } from './events/new_hello';
～省略～
export class HelloModule extends BaseModule {
	～省略～
	public constructor() {
		～省略～
		this.events.register(NewHelloEvent, new NewHelloEvent(this.name));
	}
	～省略～
}

[修正内容]
インポートに作成した new_hello を追加
コンストラクタで作成したnew_helloイベントを登録

編集内容：インポートの追加とイベントの登録

3. コマンドからイベントの呼び出し

イベントを利用するためにsrc/app/modules/hello/commands/create_hello_command.tsを以下のように修正します。

～省略～
import { NewHelloEvent } from '../events/new_hello';
～省略～
export class CreateHelloCommand extends BaseCommand {
	～省略～
	public async execute(context: CommandExecuteContext<Params>): Promise<void> {
		～省略～
		// 4. Emit a "New Hello" event
		const newHelloEvent = this.events.get(NewHelloEvent);
		newHelloEvent.add(context, {
			senderAddress: context.transaction.senderAddress,
			message: context.params.message
		},[context.transaction.senderAddress]);
	}
}

[修正内容]
インポートに作成した new_hello を追加
executeメソッドからnew_helloイベントのaddを呼び出し
※add呼び出し時のパラメータには作成したイベントのスキーマに合わせて送信者のアドレスおよびメッセージを設定。

編集内容：インポートの追加

編集内容：executeメソッドの編集

プラグインの作成

1. プラグインの作成

参考公式ドキュメント：How to create a plugin

以下のコマンドを実行します。

cd ~/hello
lisk generate:plugin helloInfo

もしコマンドを実行後、コンソールに「? Overwrite .liskrc.json?」というメッセージが表示されたら y と入力後、ENTERキーを押しましょう。

そうすると、src/app/plugins/hello_info/hello_info_plugin.tsが作成されます。
また、src/app/plugins.tsが以下のように書き換わります。

/* eslint-disable @typescript-eslint/no-empty-function */
import { DashboardPlugin } from '@liskhq/lisk-framework-dashboard-plugin';
import { Application } from 'lisk-sdk';
import { HelloInfoPlugin } from "./plugins/hello_info/hello_info_plugin";

export const registerPlugins = (app: Application): void => {
    app.registerPlugin(new DashboardPlugin());
    app.registerPlugin(new HelloInfoPlugin());
};

[変更内容]
app.registerPlugin(new HelloInfoPlugin()); の追加

2. スキーマの作成

参考公式ドキュメント：1. Defining off-chain data structures
参考公式ドキュメント：3. Configuring a Plugin

src/app/plugins/hello_info/直下にschema.tsを作成し、以下を記述します。

export const configSchema = {
	$id: '/plugins/helloInfo/config',
	type: 'object',
	properties: {
		syncInterval: {
			type: 'integer',
		},
	},
	required: ['syncInterval'],
	default: {
		syncInterval: 120000, // milliseconds
	},
};

export const offChainEventSchema = {
	$id: '/helloInfo/newHello',
	type: 'object',
	required: ['senderAddress', 'message', 'height'],
	properties: {
		senderAddress: {
			dataType: 'bytes',
			fieldNumber: 1,
		},
		message: {
			dataType: 'string',
			fieldNumber: 2,
		},
		height: {
			dataType: 'uint32',
			fieldNumber: 3,
		},
	},
};

export const counterSchema = {
	$id: '/helloInfo/counter',
	type: 'object',
	required: ['counter'],
	properties: {
		counter: {
			dataType: 'uint32',
			fieldNumber: 1,
		},
	},
};

export const heightSchema = {
	$id: '/helloInfo/height',
	type: 'object',
	required: ['height'],
	properties: {
		height: {
			dataType: 'uint32',
			fieldNumber: 1,
		},
	},
};

[configSchema]
プラグインの設定用のスキーマです。
この設定の場合、同期時間を120000ミリ秒(120秒)を初期値としています。

[offChainEventSchema]
送信者のアドレス(senderAddress)、メッセージ(message)およびコマンドが含まれたブロックの高さ(height)を設定するために使用します。

[counterSchema]
オフチェーンストアに保持するキー情報して使用します。

[heightSchema]
オフチェーンストアにイベント情報を同期する際にどのブロックの高さまで同期したかを管理するために使用します。

また、上記で作成したスキーマに合わせて src/app/plugins/hello_info/直下にtypes.tsを作成し、以下を記述します。

export interface HelloInfoPluginConfig {
    syncInterval: number;
}

export interface Event {
    senderAddress: Buffer;
    message: string;
    height: number;
}

export interface Counter {
    counter: number;
}

export interface Height {
    height: number;
}

さらに、 src/app/plugins/hello_info/直下にconstants.tsを作成し、以下を記述します。

export const DB_KEY_EVENT_INFO = Buffer.from([0]);
export const DB_LAST_COUNTER_INFO = Buffer.from([1]);
export const DB_LAST_HEIGHT_INFO = Buffer.from([2]);

この constants.ts の内容はデータベースのキーとなるものです。
上からそれぞれ、スキーマのoffChainEventSchema、counterSchema、heightSchemaに対応するものになります。
※configSchemaに関する情報はデータベースに保存しないので不要です。

コーディングする際に混乱することのないように、こういった決まった値（定数）はconstants.tsなどの定数を管理するファイルにまとめておくことをおススメします。

3. データベースの設定

参考公式ドキュメント：2. Setting up an Off-chain database

src/app/plugins/hello_info/直下にdb.tsを作成し、以下を記述します。

import { codec, db as liskDB, cryptography } from 'lisk-sdk';
import * as os from 'os';
import { join } from 'path';
import { ensureDir } from 'fs-extra';
import { offChainEventSchema, counterSchema, heightSchema } from './schema';
import { Event, Counter, Height } from './types';
import { DB_KEY_EVENT_INFO, DB_LAST_COUNTER_INFO, DB_LAST_HEIGHT_INFO } from './constants';

const { Database } = liskDB;
type KVStore = liskDB.Database;

// Returns DB's instance.
export const getDBInstance = async (
	dataPath: string,
	dbName = 'lisk-framework-helloInfo-plugin.db',
): Promise<KVStore> => {
	const dirPath = join(dataPath.replace('~', os.homedir()), 'database', dbName);
	await ensureDir(dirPath);
	return new Database(dirPath);
};

// Stores event data in the database.
export const setEventHelloInfo = async (
	db: KVStore,
	_lskAddress: Buffer,
	_message: string,
	_eventHeight: number,
	lastCounter: number,
): Promise<void> => {
	const encodedAddressInfo = codec.encode(offChainEventSchema, {
		senderAddress: _lskAddress,
		message: _message,
		height: _eventHeight,
	});
	// Creates a unique key for each event
	const dbKey = Buffer.concat([DB_KEY_EVENT_INFO, cryptography.utils.intToBuffer(lastCounter, 4)]);
	await db.set(dbKey, encodedAddressInfo);
	console.log('** Event data saved successfully in the database **');
};

// Returns event's data stored in the database.
export const getEventHelloInfo = async (db: KVStore): Promise<(Event & { id: Buffer })[]> => {
	// 1. Look for all the given key-value pairs in the database
	const stream = db.createReadStream({
		gte: Buffer.concat([DB_KEY_EVENT_INFO, Buffer.alloc(4, 0)]),
		lte: Buffer.concat([DB_KEY_EVENT_INFO, Buffer.alloc(4, 255)]),
	});
	// 2. Get event's data out of the collected stream and push it in an array.
	const results = await new Promise<(Event & { id: Buffer })[]>((resolve, reject) => {
		const events: (Event & { id: Buffer })[] = [];
		stream
			.on('data', ({ key, value }: { key: Buffer; value: Buffer }) => {
				events.push({
					...codec.decode<Event>(offChainEventSchema, value),
					id: key.subarray(DB_KEY_EVENT_INFO.length),
				});
			})
			.on('error', error => {
				reject(error);
			})
			.on('end', () => {
				resolve(events);
			});
	});
	return results;
};

// Stores lastCounter for key generation.
export const setLastCounter = async (db: KVStore, lastCounter: number): Promise<void> => {
	const encodedCounterInfo = codec.encode(counterSchema, { counter: lastCounter });
	await db.set(DB_LAST_COUNTER_INFO, encodedCounterInfo);
	console.log('** Counter saved successfully in the database **');
};

// Returns lastCounter.
export const getLastCounter = async (db: KVStore): Promise<Counter> => {
	const encodedCounterInfo = await db.get(DB_LAST_COUNTER_INFO);
	return codec.decode<Counter>(counterSchema, encodedCounterInfo);
};

// Stores height of block where hello event exists.
export const setLastEventHeight = async (db: KVStore, lastHeight: number): Promise<void> => {
	const encodedHeightInfo = codec.encode(heightSchema, { height: lastHeight });
	await db.set(DB_LAST_HEIGHT_INFO, encodedHeightInfo);
	console.log('**Height saved successfully in the database **');
};

// Returns height of block where hello event exists.
export const getLastEventHeight = async (db: KVStore): Promise<Height> => {
	const encodedHeightInfo = await db.get(DB_LAST_HEIGHT_INFO);
	return codec.decode<Height>(heightSchema, encodedHeightInfo);
};

[getDBInstanceメソッド]
データベースのインスタンスを取得します。
データベース名は "lisk-framework-helloInfo-plugin.db"です。
※ちなみに、LiskではRocksDBを使用しています。

[setEventHelloInfoメソッド]
スキーマ：offChainEventSchema に従い、イベント情報をデータベースに保存します。

[getEventHelloInfoメソッド]
データベースに保存したイベント情報を取得します。
※公式ドキュメントでは id: key.slice(…) のようにBuffer.sliceを使用していますが、使用非推奨となっているためsliceの代わりにsubarrayを使用しましょう。

[setLastCounterメソッド]
スキーマ：counterSchemaに従い、キーとなるカウンタの最大値をデータベースに保存します。

[getLastCounterメソッド]
データベースに保存したカウンタの最大値を取得します。

[setLastEventHeightメソッド]
スキーマ：heightSchema に従い、同期したイベントのブロックの高さの最大値をデータベースに保存します。

[getLastCounterメソッド]
データベースに保存したブロックの高さの最大値を取得します。

4. プラグインの編集

参考公式ドキュメント：4. Updating the Plugin Class

src/app/plugins/hello_info/hello_info_plugin.ts を以下のように修正します。

import { BasePlugin, db as liskDB, codec } from 'lisk-sdk';
import {
	getDBInstance,
	getLastCounter,
	getLastEventHeight,
	setEventHelloInfo,
	setLastCounter,
	setLastEventHeight,
} from './db';
import { configSchema } from './schema';
import { HelloInfoPluginConfig, Height, Counter } from './types';
import { newHelloEventSchema } from '../../modules/hello/events/new_hello';

export class HelloInfoPlugin extends BasePlugin<HelloInfoPluginConfig> {
	public configSchema = configSchema;
	private _pluginDB!: liskDB.Database;

	public get nodeModulePath(): string {
		return __filename;
	}

	public async load(): Promise<void> {
		// loads DB instance
		this._pluginDB = await getDBInstance(this.dataPath);
		// Syncs plugin's database after an interval.
		setInterval(() => { this._syncChainEvents(); }, this.config.syncInterval);
	}

	// eslint-disable-next-line @typescript-eslint/require-await
	public async unload(): Promise<void> {
		this._pluginDB.close();
	}

	private async _getLastCounter(): Promise<Counter> {
		try {
			const counter = await getLastCounter(this._pluginDB);
			return counter;
		} catch (error) {
			if (!(error instanceof liskDB.NotFoundError)) {
				throw error;
			}
			await setLastCounter(this._pluginDB, 0);
			return { counter: 0 };
		}
	}

	private async _getLastHeight(): Promise<Height> {
		try {
			const height = await getLastEventHeight(this._pluginDB);
			return height;
		} catch (error) {
			if (!(error instanceof liskDB.NotFoundError)) {
				throw error;
			}
			await setLastEventHeight(this._pluginDB, 0);
			return { height: 0 };
		}
	}
	
	private async _saveEventInfoToDB(parsedData: { senderAddress: Buffer; message: string }, chainHeight: number, counterValue: number): Promise<string> {
		// 1. Saves newly generated hello events to the database
		const { senderAddress, message } = parsedData;
		await setEventHelloInfo(this._pluginDB, senderAddress, message, chainHeight, counterValue);
		// 2. Saves incremented counter value
		await setLastCounter(this._pluginDB, counterValue);
		// 3. Saves last checked block's height
		await setLastEventHeight(this._pluginDB, chainHeight);
		return "Data Saved";
	}
	
	private async _syncChainEvents(): Promise<void> {
		// 1. Get the latest block height from the blockchain
		const res = await this.apiClient.invoke<{ header: { height: number } }>("chain_getLastBlock", {
		})
		// 2. Get block height stored in the database
		const heightObj = await this._getLastHeight();
		const lastStoredHeight = heightObj.height + 1;
		const { height } = res.header;
		// 3. Loop through new blocks, starting from the lastStoredHeight + 1
		for (let index = lastStoredHeight; index <= height; index += 1) {
			const result = await this.apiClient.invoke<{ data: string; height: number; module: string; name: string }[]>("chain_getEvents", {
				height: index
			});
			// 3a. Once an event is found, decode its data and pass it to the _saveEventInfoToDB() function
			const helloEvents = result.filter(e => e.module === 'hello' && e.name === 'newHello');
			for (const helloEvent of helloEvents) {
				const parsedData = codec.decode<{ senderAddress: Buffer; message: string }>(newHelloEventSchema, Buffer.from(helloEvent.data, 'hex'));
				const { counter } = await this._getLastCounter();
				await this._saveEventInfoToDB(parsedData, helloEvent.height, counter + 1);
			}
		}
		// 4. At the end of the loop, save the last checked block height in the database.
		await setLastEventHeight(this._pluginDB, height);
	}
}

[_getLastCounterメソッド]
db.ts のgetLastCounterメソッドを呼び出しカウンタの最大値を取得します。

[_getLastHeightメソッド]
db.tsのgetLastHeightメソッドを呼び出しブロックの高さの最大値を取得します。

[_saveEventInfoToDBメソッド]
イベントと同期し、イベント情報、カウンタ、ブロックの高さをデータベースに保存します。

[_syncChainEventsメソッド]
_getLastHeightメソッドで取得した前回同期したブロックの高さの最大値から同期時点のブロックの高さまに発生したイベントに対して処理を行います。
また、取得したイベントを_saveEventInfoToDBメソッドを使用して保存する際に、キーが重複しないように_getLastCounterメソッドで取得したカウンタに対して+1します。
この処理はプラグインのloadメソッドで呼び出されます。

5. プラグイン用のエンドポイントの作成

参考公式ドキュメント：5. Creating an endpoint & Testing the plugin

src/app/plugins/hello_info/ 直下にendpoint.tsを作成し、以下を記述します

import {
    BasePluginEndpoint,
    PluginEndpointContext,
    db as liskDB,
    cryptography,
} from 'lisk-sdk';
import {
    getEventHelloInfo,
} from './db';

export class Endpoint extends BasePluginEndpoint {
    private _pluginDB!: liskDB.Database;

    // Initialize the database instance here
    public init(db: liskDB.Database) {
        this._pluginDB = db;
    }

    // Returns all Sender Addresses, Hello Messages, and Block Height of the block where the Hello Event was emitted.
    public async getMessageList(_context: PluginEndpointContext): Promise<unknown[]> {
        const data: {
            ID: number;
            senderAddress: string;
            message: string;
            blockHeight;
        }[] = [];
        // 1. Get all the stored events from the database.
        const messageList = await getEventHelloInfo(this._pluginDB);
        // 2. Push them into an array for presentation.
        for (const helloMessage of messageList) {
            data.push({
                ID: helloMessage.id.readUInt32BE(0),
                senderAddress: cryptography.address.getLisk32AddressFromAddress(helloMessage['senderAddress']),
                message: helloMessage['message'],
                blockHeight: helloMessage['height'],
            })
        }
        return data;
    }
}

[initメソッド]
プラグイン用に作成したデータベース情報を設定します。
※エンドポイントの初期処理です。後述のプラグインのloadメソッドから呼び出されます。

[getMessageListメソッド]
データベースに保存されているイベント情報を取得します。
※エンドポイントのメイン処理です。このメソッドを呼び出すことでデータベースに保存されたイベント情報を取得することができます。

エンドポイントでプラグインで使用しているデータベースを使用するために src/app/plugins/hello_info/hello_info_plugin.ts を以下のように修正します。

～省略～
import { Endpoint } from './endpoint';

export class HelloInfoPlugin extends BasePlugin<HelloInfoPluginConfig> {
	～省略～
	public endpoint = new Endpoint();
	～省略～
	public async load(): Promise<void> {
		～省略～
		this.endpoint.init(this._pluginDB);
	}
	～省略～
}

[修正内容]
インポートに作成した endpoint を追加
メンバ変数：_endpoint として作成したendpointのインスタンスを設定
loadメソッドより作成したendpointのinitメソッドを呼び出し

エンドポイントの実行

ここまで出来たらビルド後にブロックチェーンアプリを実行しましょう。
アプリ実行は前回PM2の導入を行っている方は以下のコマンドです。

cd ~/hello
npm run build
pm2 start pm2_config.json

また、以下のコマンドで実行ログをしばらく眺めていると、「 **Height saved successfully in the database**」といったログが出力されるはずです。

pm2 logs

「2. スキーマの作成」で同期時間を120000ミリ秒(120秒)としていた場合は起動から2分程度で出力されるはずです。
なお、custom_config.jsonでこの同期時間を変更することもできます。

また、以下のコマンドでログファイル内をグレップして動作していることを確認してもいいと思います。

cat ~/.pm2/logs/hello-out.log | grep -i saved

動作確認が出来たら、以下のコマンドを実行します。

cd ~/hello
./bin/run endpoint:invoke helloInfo_getMessageList --pretty

初回はおそらく [] と出力されると思います。
なので、ダッシュボードプラグイン(http://localhost:4005)でcreateHelloコマンドを実行後にやってみましょう。
コマンドの実行については以下を参照。

コマンドを何度か実行後にやってみるとすべての情報が表示されると思います。
どうですか？出力されましたか？

全イベントが取得された！やったね！

おわりに

今回はイベントとプラグインの作成をしましたがいかがでしたでしょうか？
プラグインの作成周りはコード量は多いように思いますが、やってることはDBからの情報取得とDBへの情報設定という単純なものを3種類作っているだけなので実は書いてみるとたいしたことはありませんｗ

そして！サイドチェーンにしない場合はここまでで一通り終了です。
モジュールだけ作成したり、プラグインを作成したり、ぼくのかんがえたさいきょーのブロックチェーンを作ることもできるはずですｗ

なお、サイドチェーンにする場合は公式ドキュメント「How to register a sidechain」を参照してください。（Lisk自体のノードと絡むのでこのシリーズでやるかは未定。）

ということでお疲れさまでした！

おまけ：今回触ったソースの全内容

src/app/plugins.ts

/* eslint-disable @typescript-eslint/no-empty-function */
import { DashboardPlugin } from '@liskhq/lisk-framework-dashboard-plugin';
import { Application } from 'lisk-sdk';
import { HelloInfoPlugin } from "./plugins/hello_info/hello_info_plugin";

export const registerPlugins = (app: Application): void => {
    app.registerPlugin(new DashboardPlugin());
    app.registerPlugin(new HelloInfoPlugin());
};

src/app/modules/hello/module.ts

/* eslint-disable class-methods-use-this */
/* eslint-disable @typescript-eslint/member-ordering */

import { validator } from '@liskhq/lisk-validator';
import {
	BaseModule,
	ModuleInitArgs,
	ModuleMetadata,
	// InsertAssetContext,
	// BlockVerifyContext,
	TransactionVerifyContext,
	VerificationResult,
	// TransactionExecuteContext,
	// GenesisBlockExecuteContext,
	// BlockExecuteContext,
	// BlockAfterExecuteContext,
	VerifyStatus,
	utils,
} from 'lisk-sdk';
import { CreateHelloCommand } from './commands/create_hello_command';
import { HelloEndpoint } from './endpoint';
import { HelloMethod } from './method';
import { configSchema, getHelloRequestSchema, getHelloResponseSchema } from './schema';
import { MessageStore } from './stores/message';
import { ModuleConfigJSON } from './types'
import { NewHelloEvent } from './events/new_hello';

export const defaultConfig = {
	maxMessageLength: 256,
	minMessageLength: 3,
	blacklist: ['illegalWord1'],
};

export class HelloModule extends BaseModule {
	public endpoint = new HelloEndpoint(this.stores, this.offchainStores);
	public method = new HelloMethod(this.stores, this.events);
	public commands = [new CreateHelloCommand(this.stores, this.events)];

	public constructor() {
		super();
		// registeration of stores and events
		this.stores.register(MessageStore, new MessageStore(this.name, 0));
		this.events.register(NewHelloEvent, new NewHelloEvent(this.name));
	}

	public metadata(): ModuleMetadata {
		return {
			endpoints: [
				{
					name: this.endpoint.getHello.name,
					request: getHelloRequestSchema,
					response: getHelloResponseSchema,
				},
			],
			commands: this.commands.map(command => ({
				name: command.name,
				params: command.schema,
			})),
			events: this.events.values().map(v => ({
				name: v.name,
				data: v.schema,
			})),
			assets: [],
			stores: [],
		};
	}

	// Lifecycle hooks
	public async init(args: ModuleInitArgs): Promise<void> {
		// Get the module config defined in the config.json of the node
		const { moduleConfig } = args;
		// Overwrite the default module config with values from config.json, if set
		const config = utils.objects.mergeDeep({}, defaultConfig, moduleConfig) as ModuleConfigJSON;
		// Validate the config with the config schema
		validator.validate<ModuleConfigJSON>(configSchema, config);
		// Call the command init() method with config as parameter
		this.commands[0].init(config).catch(err => {
			console.log('Error: ', err);
		});
	}

	// public async insertAssets(_context: InsertAssetContext) {
	// 	// initialize block generation, add asset
	// }

	// public async verifyAssets(_context: BlockVerifyContext): Promise<void> {
	// 	// verify block
	// }

	// Lifecycle hooks
	public async verifyTransaction(context: TransactionVerifyContext): Promise<VerificationResult> {
		// verify transaction will be called multiple times in the transaction pool
		context.logger.info('TX VERIFICATION');
		return { status: VerifyStatus.OK };
	}

	// public async beforeCommandExecute(_context: TransactionExecuteContext): Promise<void> {
	// }

	// public async afterCommandExecute(_context: TransactionExecuteContext): Promise<void> {

	// }
	// public async initGenesisState(_context: GenesisBlockExecuteContext): Promise<void> {

	// }

	// public async finalizeGenesisState(_context: GenesisBlockExecuteContext): Promise<void> {

	// }

	// public async beforeTransactionsExecute(_context: BlockExecuteContext): Promise<void> {

	// }

	// public async afterTransactionsExecute(_context: BlockAfterExecuteContext): Promise<void> {

	// }
}

src/app/modules/hello/commands/create_hello_command.ts

/* eslint-disable class-methods-use-this */
import {
	BaseCommand,
	CommandVerifyContext,
	CommandExecuteContext,
	VerificationResult,
	VerifyStatus,
} from 'lisk-sdk';
import { createHelloSchema } from '../schema';
import { MessageStore } from '../stores/message';
import { ModuleConfig } from '../types';
import { NewHelloEvent } from '../events/new_hello';

interface Params {
	message: string;
}

export class CreateHelloCommand extends BaseCommand {
	public schema = createHelloSchema;
	private _blacklist!: string[];

	public async init(config: ModuleConfig): Promise<void> {
		// Set _blacklist to the value of the blacklist defined in the module config
		this._blacklist = config.blacklist;
		// Set the max message length to the value defined in the module config
		this.schema.properties.message.maxLength = config.maxMessageLength;
		// Set the min message length to the value defined in the module config
		this.schema.properties.message.minLength = config.minMessageLength;
	}

	// eslint-disable-next-line @typescript-eslint/require-await
	public async verify(context: CommandVerifyContext<Params>): Promise<VerificationResult> {
		let validation: VerificationResult;
		const wordList = context.params.message.split(' ');
		const found = this._blacklist.filter(value => wordList.includes(value));
		if (found.length > 0) {
			context.logger.info('==== FOUND: Message contains a blacklisted word ====');
			throw new Error(`Illegal word in hello message: ${found.toString()}`);
		} else {
			context.logger.info('==== NOT FOUND: Message contains no blacklisted words ====');
			validation = {
				status: VerifyStatus.OK,
			};
		}
		return validation;
	}

	public async execute(context: CommandExecuteContext<Params>): Promise<void> {
		// 1. Get account data of the sender of the Hello transaction.
		const { senderAddress } = context.transaction;
		// 2. Get message store.
		const messageSubstore = this.stores.get(MessageStore);
		// 3. Save the Hello message to the message store, using the senderAddress as key, and the message as value.
		await messageSubstore.set(context, senderAddress, {
			message: context.params.message,
		});
		// 4. Emit a "New Hello" event
		const newHelloEvent = this.events.get(NewHelloEvent);
		newHelloEvent.add(context, {
			senderAddress: context.transaction.senderAddress,
			message: context.params.message
		},[context.transaction.senderAddress]);
	}
}

src/app/modules/hello/events/new_hello.ts

import { BaseEvent } from 'lisk-sdk';

export interface NewHelloEventData {
	senderAddress: Buffer;
	message: string;
}

export const newHelloEventSchema = {
	$id: '/hello/events/new_hello',
	type: 'object',
	required: ['senderAddress', 'message'],
	properties: {
		senderAddress: {
			dataType: 'bytes',
			fieldNumber: 1,
		},
		message: {
			dataType: 'string',
			fieldNumber: 2,
		},
	},
};

export class NewHelloEvent extends BaseEvent<NewHelloEventData> {
	public schema = newHelloEventSchema;
}

src/app/plugins/hello_info/constants.ts

export const DB_KEY_EVENT_INFO = Buffer.from([0]);
export const DB_LAST_COUNTER_INFO = Buffer.from([1]);
export const DB_LAST_HEIGHT_INFO = Buffer.from([2]);

src/app/plugins/hello_info/db.ts

import { codec, db as liskDB, cryptography } from 'lisk-sdk';
import * as os from 'os';
import { join } from 'path';
import { ensureDir } from 'fs-extra';
import { offChainEventSchema, counterSchema, heightSchema } from './schema';
import { Event, Counter, Height } from './types';
import { DB_KEY_EVENT_INFO, DB_LAST_COUNTER_INFO, DB_LAST_HEIGHT_INFO } from './constants';

const { Database } = liskDB;
type KVStore = liskDB.Database;

// Returns DB's instance.
export const getDBInstance = async (
	dataPath: string,
	dbName = 'lisk-framework-helloInfo-plugin.db',
): Promise<KVStore> => {
	const dirPath = join(dataPath.replace('~', os.homedir()), 'database', dbName);
	await ensureDir(dirPath);
	return new Database(dirPath);
};

// Stores event data in the database.
export const setEventHelloInfo = async (
	db: KVStore,
	_lskAddress: Buffer,
	_message: string,
	_eventHeight: number,
	lastCounter: number,
): Promise<void> => {
	const encodedAddressInfo = codec.encode(offChainEventSchema, {
		senderAddress: _lskAddress,
		message: _message,
		height: _eventHeight,
	});
	// Creates a unique key for each event
	const dbKey = Buffer.concat([DB_KEY_EVENT_INFO, cryptography.utils.intToBuffer(lastCounter, 4)]);
	await db.set(dbKey, encodedAddressInfo);
	console.log('** Event data saved successfully in the database **');
};

// Returns event's data stored in the database.
export const getEventHelloInfo = async (db: KVStore): Promise<(Event & { id: Buffer })[]> => {
	// 1. Look for all the given key-value pairs in the database
	const stream = db.createReadStream({
		gte: Buffer.concat([DB_KEY_EVENT_INFO, Buffer.alloc(4, 0)]),
		lte: Buffer.concat([DB_KEY_EVENT_INFO, Buffer.alloc(4, 255)]),
	});
	// 2. Get event's data out of the collected stream and push it in an array.
	const results = await new Promise<(Event & { id: Buffer })[]>((resolve, reject) => {
		const events: (Event & { id: Buffer })[] = [];
		stream
			.on('data', ({ key, value }: { key: Buffer; value: Buffer }) => {
				events.push({
					...codec.decode<Event>(offChainEventSchema, value),
					id: key.subarray(DB_KEY_EVENT_INFO.length),
				});
			})
			.on('error', error => {
				reject(error);
			})
			.on('end', () => {
				resolve(events);
			});
	});
	return results;
};

// Stores lastCounter for key generation.
export const setLastCounter = async (db: KVStore, lastCounter: number): Promise<void> => {
	const encodedCounterInfo = codec.encode(counterSchema, { counter: lastCounter });
	await db.set(DB_LAST_COUNTER_INFO, encodedCounterInfo);
	console.log('** Counter saved successfully in the database **');
};

// Returns lastCounter.
export const getLastCounter = async (db: KVStore): Promise<Counter> => {
	const encodedCounterInfo = await db.get(DB_LAST_COUNTER_INFO);
	return codec.decode<Counter>(counterSchema, encodedCounterInfo);
};

// Stores height of block where hello event exists.
export const setLastEventHeight = async (db: KVStore, lastHeight: number): Promise<void> => {
	const encodedHeightInfo = codec.encode(heightSchema, { height: lastHeight });
	await db.set(DB_LAST_HEIGHT_INFO, encodedHeightInfo);
	console.log('**Height saved successfully in the database **');
};

// Returns height of block where hello event exists.
export const getLastEventHeight = async (db: KVStore): Promise<Height> => {
	const encodedHeightInfo = await db.get(DB_LAST_HEIGHT_INFO);
	return codec.decode<Height>(heightSchema, encodedHeightInfo);
};

src/app/plugins/hello_info/endpoint.ts

import {
    BasePluginEndpoint,
    PluginEndpointContext,
    db as liskDB,
    cryptography,
} from 'lisk-sdk';
import {
    getEventHelloInfo,
} from './db';

export class Endpoint extends BasePluginEndpoint {
    private _pluginDB!: liskDB.Database;

    // Initialize the database instance here
    public init(db: liskDB.Database) {
        this._pluginDB = db;
    }

    // Returns all Sender Addresses, Hello Messages, and Block Height of the block where the Hello Event was emitted.
    public async getMessageList(_context: PluginEndpointContext): Promise<unknown[]> {
        const data: {
            ID: number;
            senderAddress: string;
            message: string;
            blockHeight;
        }[] = [];
        // 1. Get all the stored events from the database.
        const messageList = await getEventHelloInfo(this._pluginDB);
        // 2. Push them into an array for presentation.
        for (const helloMessage of messageList) {
            data.push({
                ID: helloMessage.id.readUInt32BE(0),
                senderAddress: cryptography.address.getLisk32AddressFromAddress(helloMessage['senderAddress']),
                message: helloMessage['message'],
                blockHeight: helloMessage['height'],
            })
        }
        return data;
    }
}

src/app/plugins/hello_info/hello_info_plugin.ts

import { BasePlugin, db as liskDB, codec } from 'lisk-sdk';
import {
	getDBInstance,
	getLastCounter,
	getLastEventHeight,
	setEventHelloInfo,
	setLastCounter,
	setLastEventHeight,
} from './db';
import { configSchema } from './schema';
import { HelloInfoPluginConfig, Height, Counter } from './types';
import { newHelloEventSchema } from '../../modules/hello/events/new_hello';
import { Endpoint } from './endpoint';

export class HelloInfoPlugin extends BasePlugin<HelloInfoPluginConfig> {
	public configSchema = configSchema;
	public endpoint = new Endpoint();
	private _pluginDB!: liskDB.Database;

	public get nodeModulePath(): string {
		return __filename;
	}

	public async load(): Promise<void> {
		// loads DB instance
		this._pluginDB = await getDBInstance(this.dataPath);
		// Syncs plugin's database after an interval.
		setInterval(() => { this._syncChainEvents(); }, this.config.syncInterval);
		this.endpoint.init(this._pluginDB);
	}

	// eslint-disable-next-line @typescript-eslint/require-await
	public async unload(): Promise<void> {
		this._pluginDB.close();
	}

	private async _getLastCounter(): Promise<Counter> {
		try {
			const counter = await getLastCounter(this._pluginDB);
			return counter;
		} catch (error) {
			if (!(error instanceof liskDB.NotFoundError)) {
				throw error;
			}
			await setLastCounter(this._pluginDB, 0);
			return { counter: 0 };
		}
	}

	private async _getLastHeight(): Promise<Height> {
		try {
			const height = await getLastEventHeight(this._pluginDB);
			return height;
		} catch (error) {
			if (!(error instanceof liskDB.NotFoundError)) {
				throw error;
			}
			await setLastEventHeight(this._pluginDB, 0);
			return { height: 0 };
		}
	}
	
	private async _saveEventInfoToDB(parsedData: { senderAddress: Buffer; message: string }, chainHeight: number, counterValue: number): Promise<string> {
		// 1. Saves newly generated hello events to the database
		const { senderAddress, message } = parsedData;
		await setEventHelloInfo(this._pluginDB, senderAddress, message, chainHeight, counterValue);
		// 2. Saves incremented counter value
		await setLastCounter(this._pluginDB, counterValue);
		// 3. Saves last checked block's height
		await setLastEventHeight(this._pluginDB, chainHeight);
		return "Data Saved";
	}
	
	private async _syncChainEvents(): Promise<void> {
		// 1. Get the latest block height from the blockchain
		const res = await this.apiClient.invoke<{ header: { height: number } }>("chain_getLastBlock", {
		})
		// 2. Get block height stored in the database
		const heightObj = await this._getLastHeight();
		const lastStoredHeight = heightObj.height + 1;
		const { height } = res.header;
		// 3. Loop through new blocks, starting from the lastStoredHeight + 1
		for (let index = lastStoredHeight; index <= height; index += 1) {
			const result = await this.apiClient.invoke<{ data: string; height: number; module: string; name: string }[]>("chain_getEvents", {
				height: index
			});
			// 3a. Once an event is found, decode its data and pass it to the _saveEventInfoToDB() function
			const helloEvents = result.filter(e => e.module === 'hello' && e.name === 'newHello');
			for (const helloEvent of helloEvents) {
				const parsedData = codec.decode<{ senderAddress: Buffer; message: string }>(newHelloEventSchema, Buffer.from(helloEvent.data, 'hex'));
				const { counter } = await this._getLastCounter();
				await this._saveEventInfoToDB(parsedData, helloEvent.height, counter + 1);
			}
		}
		// 4. At the end of the loop, save the last checked block height in the database.
		await setLastEventHeight(this._pluginDB, height);
	}
}

src/app/plugins/hello_info/schema.ts

export const configSchema = {
	$id: '/plugins/helloInfo/config',
	type: 'object',
	properties: {
		syncInterval: {
			type: 'integer',
		},
	},
	required: ['syncInterval'],
	default: {
		syncInterval: 120000, // milliseconds
	},
};

export const offChainEventSchema = {
	$id: '/helloInfo/newHello',
	type: 'object',
	required: ['senderAddress', 'message', 'height'],
	properties: {
		senderAddress: {
			dataType: 'bytes',
			fieldNumber: 1,
		},
		message: {
			dataType: 'string',
			fieldNumber: 2,
		},
		height: {
			dataType: 'uint32',
			fieldNumber: 3,
		},
	},
};

export const counterSchema = {
	$id: '/helloInfo/counter',
	type: 'object',
	required: ['counter'],
	properties: {
		counter: {
			dataType: 'uint32',
			fieldNumber: 1,
		},
	},
};

export const heightSchema = {
	$id: '/helloInfo/height',
	type: 'object',
	required: ['height'],
	properties: {
		height: {
			dataType: 'uint32',
			fieldNumber: 1,
		},
	},
};

src/app/plugins/hello_info/types.ts

export interface HelloInfoPluginConfig {
    syncInterval: number;
}

export interface Event {
    senderAddress: Buffer;
    message: string;
    height: number;
}

export interface Counter {
    counter: number;
}

export interface Height {
    height: number;
}