Merge pull request #1946 from LiskHQ/1942-emit-an-event-after-index-s…

…tatus-update

Emit an event after index status update

docs/api/websocket_subscribe_api.md

@@ -31,6 +31,8 @@ Lisk Service leverages the two-way communication approach by utilizing the WebSo
     - [Payload](#payload-6)
   - [`update.metadata`](#updatemetadata)
     - [Payload](#payload-7)
+  - [`update.index.status`](#updateindexstatus)
+    - [Payload](#payload-8)
 
 ## Access paths and compatibility
 
@@ -368,3 +370,26 @@ Updates about recent metadata changes.
   "betanet": ["Lisk"],
 }
 ```
+
+## `update.index.status`
+
+Updates about index status changes.
+
+### Payload
+
+```jsonc
+{
+  "data": {
+    "genesisHeight": 0,
+    "lastBlockHeight": 18779,
+    "lastIndexedBlockHeight": 13955,
+    "chainLength": 18780,
+    "numBlocksIndexed": 13956,
+    "percentageIndexed": 74.31,
+    "isIndexingInProgress": false
+  },
+  "meta": {
+    "lastUpdate": 1700848735
+  }
+}
+```

framework/bin/socket_io_subscribe_client.js

@@ -76,3 +76,4 @@ subscribe('update.round');
 subscribe('update.generators');
 subscribe('update.fee_estimates');
 subscribe('update.metadata');
+subscribe('update.index.status');

services/blockchain-indexer/events/blockchain.js

@@ -172,4 +172,15 @@ module.exports = [
 			Signals.get('blockIndexReady').add(indexStatusListener);
 		},
 	},
+	{
+		name: 'update.index.status',
+		description: 'Emit index status updates.',
+		controller: callback => {
+			const indexStatusUpdateListener = async payload => {
+				logger.debug("Dispatching 'update.index.status' event over websocket");
+				callback(payload);
+			};
+			Signals.get('updateIndexStatus').add(indexStatusUpdateListener);
+		},
+	},
 ];

services/blockchain-indexer/shared/dataService/indexStatus.js

@@ -21,19 +21,6 @@ let indexStatsCache = {};
 let isIndexingInProgress = false;
 let lastUpdate = getCurrentTimestamp();
 
-const indexStatUpdateListener = _indexStats => {
-	indexStatsCache = _indexStats;
-	lastUpdate = getCurrentTimestamp();
-};
-
-const indexingProgressListener = numJobsInProgress => {
-	isIndexingInProgress = numJobsInProgress > 0;
-	lastUpdate = getCurrentTimestamp();
-};
-
-Signals.get('indexStatUpdate').add(indexStatUpdateListener);
-Signals.get('numJobsInProgressUpdate').add(indexingProgressListener);
-
 const getIndexStatus = async () => {
 	const {
 		currentChainHeight,
@@ -60,9 +47,29 @@ const getIndexStatus = async () => {
 	};
 };
 
+const indexStatUpdateListener = async _indexStats => {
+	indexStatsCache = _indexStats;
+	lastUpdate = getCurrentTimestamp();
+
+	const indexStatus = await getIndexStatus();
+	Signals.get('updateIndexStatus').dispatch(indexStatus);
+};
+
+const indexingProgressListener = numJobsInProgress => {
+	isIndexingInProgress = numJobsInProgress > 0;
+	lastUpdate = getCurrentTimestamp();
+};
+
+Signals.get('indexStatUpdate').add(indexStatUpdateListener);
+Signals.get('numJobsInProgressUpdate').add(indexingProgressListener);
+
 const isBlockchainFullyIndexed = () => Number(indexStatsCache.percentage) === 100;
 
 module.exports = {
 	getIndexStatus,
 	isBlockchainFullyIndexed,
+
+	// Testing
+	indexStatUpdateListener,
+	indexingProgressListener,
 };

services/blockchain-indexer/tests/unit/shared/dataservice/indexStatus.test.js

@@ -0,0 +1,64 @@
+/*
+ * LiskHQ/lisk-service
+ * Copyright © 2023 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.
+ *
+ */
+const {
+	indexStatUpdateListener,
+	indexingProgressListener,
+	getIndexStatus,
+} = require('../../../../shared/dataService/indexStatus');
+
+describe('indexStatUpdateListener', () => {
+	it('should update index stats', async () => {
+		const mockIndexStats = {
+			genesisHeight: 0,
+			lastBlockHeight: 18779,
+			lastIndexedBlockHeight: 13955,
+			chainLength: 18780,
+			numBlocksIndexed: 13956,
+			percentageIndexed: 74.31,
+			isIndexingInProgress: false,
+		};
+
+		// Call the function
+		await indexStatUpdateListener(mockIndexStats);
+
+		// Assert the result
+		const indexStatus = await getIndexStatus();
+		// eslint-disable-next-line no-restricted-syntax, guard-for-in
+		for (const key in indexStatus) {
+			expect(mockIndexStats[key]).toEqual(indexStatus.data[key]);
+		}
+	});
+});
+
+describe('indexingProgressListener', () => {
+	it('should update indexing in progress status', async () => {
+		// Call the function
+		indexingProgressListener(2); // Assuming 2 jobs in progress
+
+		// Assert the result
+		const indexStatus = await getIndexStatus();
+		expect(indexStatus.data.isIndexingInProgress).toBe(true);
+	});
+
+	it('should update indexing in progress status to false when no jobs are in progress', async () => {
+		// Call the function
+		indexingProgressListener(0); // No jobs in progress
+
+		// Assert the result
+		const indexStatus = await getIndexStatus();
+		expect(indexStatus.data.isIndexingInProgress).toBe(false);
+	});
+});

services/gateway/app.js

@@ -38,6 +38,7 @@ const { definition: blocksDefinition } = require('./sources/version3/blocks');
 const { definition: feesDefinition } = require('./sources/version3/fees');
 const { definition: generatorsDefinition } = require('./sources/version3/generators');
 const { definition: transactionsDefinition } = require('./sources/version3/transactions');
+const { definition: indexStatusUpdateDefinition } = require('./sources/version3/indexStatus');
 
 const { host, port } = config;
 
@@ -182,6 +183,8 @@ tempApp.run().then(async () => {
 			'update.fee_estimates': payload =>
 				sendSocketIoEvent('update.fee_estimates', mapper(payload, feesDefinition)),
 			'metadata.change': payload => sendSocketIoEvent('update.metadata', payload),
+			'update.index.status': payload =>
+				sendSocketIoEvent('update.index.status', mapper(payload, indexStatusUpdateDefinition)),
 		},
 		dependencies: config.brokerDependencies,
 	};

services/gateway/shared/moleculer-io/index.js

@@ -443,7 +443,10 @@ function makeHandler(svc, handlerItem) {
 				});
 				svc.logger.info(`   <= ${chalk.green.bold('Success')} ${action}`);
 
-				return addJsonRpcEnvelope(id, res);
+				const output = addJsonRpcEnvelope(id, res);
+
+				if (respond !== undefined) respond(output);
+				else return output;
 			} catch (err) {
 				if (
 					svc.settings.log4XXResponses ||