はじめに

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

前回の記事はこちら

イベントの作成

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;
}

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));
	}
	～省略～
}

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]);
	}
}

プラグインの作成

1. プラグインの作成

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

cd ~/hello
lisk generate:plugin helloInfo

そうすると、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());
};

2. スキーマの作成

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;
}

さらに、 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]);

3. データベースの設定

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);
};

4. プラグインの編集

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);
	}
}

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

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 { Endpoint } from './endpoint';

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

エンドポイントの実行

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

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

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

pm2 logs

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

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

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

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

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

おわりに

今回はイベントとプラグインの作成をしましたがいかがでしたでしょうか？
プラグインの作成周りはコード量は多いように思いますが、やってることは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;
}