Fix export microservice (#2016)

* ⚡ Allow slack in index readiness when scheduling history export jobs

* ⚡ Do not re-schedule duplicate jobs

* 📝 Improve logging

* 🐛 Fix index readiness check logic for export job scheduling

* 🐎 Optimize code

* 🔨 Refactor code. Prefer early exit from the loop

* 🐛 Fix queue config and initialization

* 🔨 Automatically se-schedule a job if it timesout

* 🔧 Add microservice dependencies for export microservice

* 🐎 Optimize genesis asset query

* 🔨 Fix broken unit tests

* ✔️ Fix unit tests

* 🚨 Add new unit tests

* 👌 Apply review suggestion

* 🔧 Limit indexing pace to assist in app node performance

* 🐎 Use lighter endpoint invocations to verify account existence

* 🔨 Locally cache genesis token assets at init

* ✔️ Fix unit tests

* 🔨 Clear any stale intervals

* 🎨 Add logs

* 🔧 Revert ratelimiting on the indexing jobs

* 🔨 Increase query payload size

* ✅ Add unit tests

* 🐛 Add prefix handling when extracting transactionID from event topics

* 🚨 Add more unit tests

* 📝 Fix swagger docs

---------

Co-authored-by: nagdahimanshu <himanshu.nagda@lightcurve.io>

services/blockchain-connector/methods/token.js

@@ -22,6 +22,7 @@ const {
 	getTokenBalances,
 	getTokenInitializationFees,
 	tokenHasEscrowAccount,
+	getTokenBalanceAtGenesis,
 } = require('../shared/sdk');
 
 module.exports = [
@@ -84,4 +85,11 @@ module.exports = [
 		controller: async () => getTokenInitializationFees(),
 		params: {},
 	},
+	{
+		name: 'getTokenBalanceAtGenesis',
+		controller: async ({ address }) => getTokenBalanceAtGenesis(address),
+		params: {
+			address: { optional: false, type: 'string' },
+		},
+	},
 ];

services/blockchain-connector/shared/sdk/index.js

@@ -65,6 +65,7 @@ const {
 	getTotalSupply,
 	getTokenInitializationFees,
 	updateTokenInfo,
+	getTokenBalanceAtGenesis,
 } = require('./token');
 
 const {
@@ -178,7 +179,7 @@ module.exports = {
 	dryRunTransaction,
 	formatTransaction,
 
-	// Tokens
+	// Token
 	tokenHasUserAccount,
 	tokenHasEscrowAccount,
 	getTokenBalance,
@@ -187,6 +188,7 @@ module.exports = {
 	getSupportedTokens,
 	getTotalSupply,
 	getTokenInitializationFees,
+	getTokenBalanceAtGenesis,
 
 	// PoS
 	getAllPosValidators,

services/blockchain-connector/shared/sdk/token.js

@@ -15,6 +15,8 @@
  */
 const { invokeEndpoint } = require('./client');
 const { isMainchain } = require('./interoperability');
+const { getGenesisAssetByModule } = require('./genesisBlock');
+const { MODULE_NAME_TOKEN } = require('./constants/names');
 
 let escrowedAmounts;
 let supportedTokens;
@@ -73,6 +75,20 @@ const updateTokenInfo = async () => {
 	totalSupply = await getTotalSupply(true);
 };
 
+const getTokenBalanceAtGenesis = async address => {
+	const MODULE_TOKEN_SUBSTORE_USER = 'userSubstore';
+
+	const tokenModuleGenesisAssets = await getGenesisAssetByModule({
+		module: MODULE_NAME_TOKEN,
+		subStore: MODULE_TOKEN_SUBSTORE_USER,
+	});
+
+	const balancesAtGenesis = tokenModuleGenesisAssets[MODULE_TOKEN_SUBSTORE_USER];
+	const balancesByAddress = balancesAtGenesis.find(e => e.address === address);
+
+	return balancesByAddress;
+};
+
 module.exports = {
 	tokenHasUserAccount: hasUserAccount,
 	tokenHasEscrowAccount: hasEscrowAccount,
@@ -83,4 +99,5 @@ module.exports = {
 	getTotalSupply,
 	getTokenInitializationFees,
 	updateTokenInfo,
+	getTokenBalanceAtGenesis,
 };

services/blockchain-indexer/methods/dataService/modules/token.js

@@ -61,7 +61,8 @@ module.exports = [
 			address: { optional: true, type: 'string', pattern: regex.ADDRESS_LISK32 },
 			publicKey: { optional: true, type: 'string', pattern: regex.PUBLIC_KEY },
 			name: { optional: true, type: 'string', pattern: regex.NAME },
-			tokenID: { optional: false, type: 'string', pattern: regex.TOKEN_ID },
+			// Set tokenID as optional in indexer because export microservice needs it to be optional. Should remain mandatory everywhere else.
+			tokenID: { optional: true, type: 'string', pattern: regex.TOKEN_ID },
 		},
 	},
 	{

services/blockchain-indexer/shared/dataService/business/token/accountExists.js

@@ -17,6 +17,7 @@ const { requestConnector } = require('../../../utils/request');
 const { getAddressByName } = require('../../utils/validator');
 
 const { getLisk32AddressFromPublicKey } = require('../../../utils/account');
+const { getAvailableTokenIDs } = require('./availableIDs');
 
 const tokenHasUserAccount = async params => {
 	const response = {
@@ -26,7 +27,6 @@ const tokenHasUserAccount = async params => {
 		meta: {},
 	};
 
-	const { tokenID } = params;
 	let { address } = params;
 
 	if (!address && params.name) {
@@ -39,11 +39,28 @@ const tokenHasUserAccount = async params => {
 
 	// Check existence if address found. Return false otherwise
 	if (address) {
-		const { exists: isExists } = await requestConnector('tokenHasUserAccount', {
-			address,
-			tokenID,
-		});
-		response.data.isExists = isExists;
+		const tokenIDs = [];
+
+		if (params.tokenID) {
+			tokenIDs.push(params.tokenID);
+		} else {
+			// Logic introduced to support the export microservice
+			const result = await getAvailableTokenIDs();
+			tokenIDs.push(...result.data.tokenIDs);
+		}
+
+		// eslint-disable-next-line no-restricted-syntax
+		for (const tokenID of tokenIDs) {
+			const { exists: isExists } = await requestConnector('tokenHasUserAccount', {
+				address,
+				tokenID,
+			});
+
+			if (isExists) {
+				response.data.isExists = isExists;
+				break;
+			}
+		}
 	}
 
 	return response;

services/blockchain-indexer/shared/dataService/business/token/availableIDs.js

@@ -26,7 +26,7 @@ const MYSQL_ENDPOINT = config.endpoints.mysqlReplica;
 
 const getAccountBalancesTable = () => getTableInstance(accountBalancesTableSchema, MYSQL_ENDPOINT);
 
-const getAvailableTokenIDs = async params => {
+const getAvailableTokenIDs = async (params = {}) => {
 	const response = {
 		data: {},
 		meta: {},

services/blockchain-indexer/shared/indexer/genesisBlock.js

@@ -197,6 +197,7 @@ const indexPosModuleAssets = async dbTrx => {
 };
 
 const indexGenesisBlockAssets = async dbTrx => {
+	clearTimeout(intervalTimeout);
 	logger.info('Starting to index the genesis assets.');
 	intervalTimeout = setInterval(
 		() => logger.info('Genesis assets indexing still in progress...'),

services/blockchain-indexer/shared/utils/transactions.js

@@ -28,7 +28,9 @@ const getTransactionExecutionStatus = (tx, events) => {
 		e => e.topics.includes(EVENT_TOPIC_PREFIX.TX_ID.concat(tx.id)) || e.topics.includes(tx.id),
 	);
 	if (!txExecResultEvent)
-		throw Error(`Event unavailable to determine execution status for transaction: ${tx.id}.`);
+		throw Error(
+			`Event unavailable to determine execution status for transaction: ${tx.id}.\nEnsure that you have set 'system.keepEventsForHeights: -1' in your node config before syncing it with the network.`,
+		);
 
 	return txExecResultEvent.data.success ? TRANSACTION_STATUS.SUCCESSFUL : TRANSACTION_STATUS.FAILED;
 };

services/blockchain-indexer/tests/unit/shared/dataservice/business/token/accountExists.test.js

@@ -0,0 +1,156 @@
+/*
+ * LiskHQ/lisk-service
+ * Copyright © 2024 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.
+ *
+ */
+
+/* eslint-disable import/no-dynamic-require */
+const { resolve } = require('path');
+
+const mockRequestFilePath = resolve(`${__dirname}/../../../../../../shared/utils/request`);
+const mockTokenAvailableIDsFilePath = resolve(
+	`${__dirname}/../../../../../../shared/dataService/business/token/availableIDs`,
+);
+const mockTokenaccountExistsFilePath = resolve(
+	`${__dirname}/../../../../../../shared/dataService/business/token/accountExists`,
+);
+const mockValidatorUtilsPath = resolve(
+	`${__dirname}/../../../../../../shared/dataService/utils/validator`,
+);
+
+beforeEach(() => jest.resetModules());
+
+jest.mock('lisk-service-framework', () => {
+	const actual = jest.requireActual('lisk-service-framework');
+	return {
+		...actual,
+		DB: {
+			...actual.DB,
+			MySQL: {
+				...actual.DB.MySQL,
+				KVStore: {
+					...actual.DB.MySQL.KVStore,
+					getKeyValueTable: jest.fn(),
+				},
+			},
+		},
+		CacheRedis: jest.fn(),
+		CacheLRU: jest.fn(),
+	};
+});
+
+describe('tokenHasUserAccount', () => {
+	const tokenID = '0000000000000000';
+	const accAddressExists = 'lskyvvam5rxyvbvofxbdfcupxetzmqxu22phm4yuo';
+	const accAddressNotExists = 'lskz23xokaxhmmkpbzdjt5agcq59qkby7bne2hwpk';
+	const name = 'testAccount';
+	const publicKey = '3972849f2ab66376a68671c10a00e8b8b67d880434cc65b04c6ed886dfa91c2c';
+
+	describe('should return isExists true when user account exists', () => {
+		it('when called with tokenID and address', async () => {
+			jest.mock(mockRequestFilePath, () => ({
+				requestConnector: jest.fn(() => ({ exists: true })),
+			}));
+
+			// Make a query to tokenHasUserAccount function
+			const { tokenHasUserAccount } = require(mockTokenaccountExistsFilePath);
+			const result = await tokenHasUserAccount({ address: accAddressExists, tokenID });
+
+			expect(result).toEqual({
+				data: {
+					isExists: true,
+				},
+				meta: {},
+			});
+		});
+
+		it('when called with tokenID and publicKey', async () => {
+			jest.mock(mockRequestFilePath, () => ({
+				requestConnector: jest.fn(() => ({ exists: true })),
+			}));
+
+			// Make a query to tokenHasUserAccount function
+			const { tokenHasUserAccount } = require(mockTokenaccountExistsFilePath);
+			const result = await tokenHasUserAccount({ publicKey, tokenID });
+			expect(result).toEqual({
+				data: {
+					isExists: true,
+				},
+				meta: {},
+			});
+		});
+
+		it('when called with tokenID and name', async () => {
+			jest.mock(mockRequestFilePath, () => ({
+				requestConnector: jest.fn(() => ({ exists: true })),
+			}));
+
+			jest.mock(mockValidatorUtilsPath);
+			const { getAddressByName } = require(mockValidatorUtilsPath);
+			getAddressByName.mockReturnValueOnce(accAddressExists);
+
+			// Make a query to tokenHasUserAccount function
+			const { tokenHasUserAccount } = require(mockTokenaccountExistsFilePath);
+			const result = await tokenHasUserAccount({ name, tokenID });
+			expect(result).toEqual({
+				data: {
+					isExists: true,
+				},
+				meta: {},
+			});
+		});
+
+		it('when called with address', async () => {
+			jest.mock(mockRequestFilePath, () => ({
+				requestConnector: jest.fn(() => ({ exists: true })),
+			}));
+
+			jest.mock(mockTokenAvailableIDsFilePath);
+			const { getAvailableTokenIDs } = require(mockTokenAvailableIDsFilePath);
+			getAvailableTokenIDs.mockReturnValueOnce({
+				data: { tokenIDs: ['0000000000000000'] },
+				meta: {},
+			});
+
+			// Make a query to tokenHasUserAccount function
+			const { tokenHasUserAccount } = require(mockTokenaccountExistsFilePath);
+			const result = await tokenHasUserAccount({ address: accAddressExists });
+
+			expect(result).toEqual({
+				data: {
+					isExists: true,
+				},
+				meta: {},
+			});
+		});
+	});
+
+	describe('should return isExists false when user account does not exists', () => {
+		it('when called with tokenID and address', async () => {
+			jest.mock(mockRequestFilePath, () => ({
+				requestConnector: jest.fn(() => ({ exists: false })),
+			}));
+
+			// Make a query to tokenHasUserAccount function
+			const { tokenHasUserAccount } = require(mockTokenaccountExistsFilePath);
+			const result = await tokenHasUserAccount({ address: accAddressNotExists, tokenID });
+
+			expect(result).toEqual({
+				data: {
+					isExists: false,
+				},
+				meta: {},
+			});
+		});
+	});
+});

services/export/app.js

@@ -14,14 +14,15 @@
  *
  */
 const path = require('path');
-const { Microservice, LoggerConfig, Logger } = require('lisk-service-framework');
+const { Signals, Microservice, LoggerConfig, Logger } = require('lisk-service-framework');
 
 const config = require('./config');
 
 LoggerConfig(config.log);
 
 const packageJson = require('./package.json');
 const { setAppContext } = require('./shared/helpers');
+const { getTokenBalancesAtGenesis } = require('./shared/transactionsExport');
 
 const logger = Logger();
 
@@ -32,6 +33,17 @@ const app = Microservice({
 	timeout: config.brokerTimeout,
 	packageJson,
 	logger: config.log,
+	events: {
+		systemNodeInfo: async payload => {
+			logger.debug("Received a 'systemNodeInfo' moleculer event from connecter.");
+			Signals.get('nodeInfo').dispatch(payload);
+		},
+		'update.index.status': async payload => {
+			logger.debug("Received a 'update.index.status' moleculer event from indexer.");
+			Signals.get('updateIndexStatus').dispatch(payload);
+		},
+	},
+	dependencies: ['connector', 'indexer', 'app-registry'],
 });
 
 setAppContext(app);
@@ -43,8 +55,9 @@ app.addJobs(path.join(__dirname, 'jobs'));
 // Run the application
 app
 	.run()
-	.then(() => {
+	.then(async () => {
 		logger.info(`Service started ${packageJson.name}.`);
+		await getTokenBalancesAtGenesis();
 	})
 	.catch(err => {
 		logger.fatal(`Failed to start service ${packageJson.name} due to: ${err.message}`);

services/export/config.js

@@ -78,7 +78,20 @@ config.excel.sheets = {
 config.queue = {
 	scheduleTransactionExport: {
 		name: 'ScheduleTransactionExportQueue',
-		concurrency: 50,
+		concurrency: 10,
+		options: {
+			defaultJobOptions: {
+				attempts: 5,
+				timeout: 5 * 60 * 1000, // millisecs
+				backoff: {
+					type: 'exponential',
+					delay: 1 * 60 * 1000, // millisecs
+				},
+				removeOnComplete: true,
+				removeOnFail: true,
+				stackTraceLimit: 0,
+			},
+		},
 	},
 	defaults: {
 		jobOptions: {