Fix export microservice

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/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/export/app.js

@@ -14,7 +14,7 @@
  *
  */
 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');
 
@@ -32,6 +32,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);

services/export/config.js

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

services/export/shared/helpers/chain.js

@@ -45,9 +45,15 @@ const getUniqueChainIDs = async txs => {
 	return Array.from(chainIDs);
 };
 
+const getBlocks = async params => requestIndexer('blocks', params);
+
+const getTransactions = async params => requestIndexer('transactions', params);
+
 module.exports = {
 	getCurrentChainID,
 	resolveReceivingChainID,
 	getNetworkStatus,
 	getUniqueChainIDs,
+	getBlocks,
+	getTransactions,
 };

services/export/shared/helpers/index.js

@@ -25,6 +25,8 @@ const {
 	resolveReceivingChainID,
 	getNetworkStatus,
 	getUniqueChainIDs,
+	getBlocks,
+	getTransactions,
 } = require('./chain');
 
 const { MODULE, COMMAND, EVENT, MODULE_SUB_STORE } = require('./constants');
@@ -48,7 +50,7 @@ const {
 	requestAppRegistry,
 } = require('./request');
 
-const { getDaysInMilliseconds, dateFromTimestamp, timeFromTimestamp } = require('./time');
+const { getToday, getDaysInMilliseconds, dateFromTimestamp, timeFromTimestamp } = require('./time');
 
 const {
 	normalizeTransactionAmount,
@@ -66,6 +68,8 @@ module.exports = {
 	resolveReceivingChainID,
 	getNetworkStatus,
 	getUniqueChainIDs,
+	getBlocks,
+	getTransactions,
 
 	MODULE,
 	COMMAND,
@@ -87,6 +91,7 @@ module.exports = {
 	requestConnector,
 	requestAppRegistry,
 
+	getToday,
 	getDaysInMilliseconds,
 	dateFromTimestamp,
 	timeFromTimestamp,

services/export/shared/helpers/ready.js

@@ -0,0 +1,78 @@
+/*
+ * 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.
+ *
+ */
+const Moment = require('moment');
+const MomentRange = require('moment-range');
+
+const { Signals, Logger } = require('lisk-service-framework');
+
+const config = require('../../config');
+
+const { requestIndexer, getToday, getBlocks } = require('.');
+
+const moment = MomentRange.extendMoment(Moment);
+const DATE_FORMAT = config.excel.dateFormat;
+
+const logger = Logger();
+
+let indexStatusCache;
+
+const getIndexStatus = async () => {
+	if (!indexStatusCache) {
+		indexStatusCache = await requestIndexer('index.status');
+
+		const updateIndexStatusListener = payload => {
+			indexStatusCache = payload;
+		};
+		Signals.get('updateIndexStatus').add(updateIndexStatusListener);
+	}
+	return indexStatusCache;
+};
+
+const checkIfIndexReadyForInterval = async interval => {
+	try {
+		// Blockchain fully indexed
+		const { data: indexStatus } = await getIndexStatus();
+		if (indexStatus.percentageIndexed === 100) return true;
+
+		// Requested history for only until yesterday and blockchain index can already serve the information
+		const [, toDate] = interval.split(':');
+		const to = moment(toDate, DATE_FORMAT).endOf('day');
+		const today = moment(getToday(), DATE_FORMAT);
+
+		if (to < today) {
+			const response = await getBlocks({ height: indexStatus.lastIndexedBlockHeight });
+			const [lastIndexedBlock] = response.data;
+			const lastIndexedBlockGeneratedTime = lastIndexedBlock.timestamp;
+
+			if (to <= moment(lastIndexedBlockGeneratedTime * 1000)) return true;
+		}
+
+		// Allow job scheduling if the last few blocks have not been indexed yet
+		if (indexStatus.chainLength - indexStatus.numBlocksIndexed <= 10) {
+			return true;
+		}
+	} catch (err) {
+		logger.warn(`Index readiness check for export job scheduling failed due to: ${err.message}`);
+		logger.debug(err.stack);
+	}
+
+	return false;
+};
+
+module.exports = {
+	getIndexStatus,
+	checkIfIndexReadyForInterval,
+};

services/export/shared/helpers/time.js

@@ -17,22 +17,28 @@ const moment = require('moment');
 
 const config = require('../../config');
 
-const DAY_IN_MILLISEC = moment().endOf('day').valueOf() - moment().startOf('day').valueOf() + 1;
-const getDaysInMilliseconds = days => days * DAY_IN_MILLISEC;
+const DATE_FORMAT = config.excel.dateFormat;
+const TIME_FORMAT = config.excel.timeFormat;
+
+const getToday = () => moment().format(DATE_FORMAT);
+
+const DAY_IN_MILLISECS = moment().endOf('day').valueOf() - moment().startOf('day').valueOf() + 1;
+const getDaysInMilliseconds = days => days * DAY_IN_MILLISECS;
 
 const momentFromTimestamp = timestamp => moment.unix(timestamp);
 
 const dateFromTimestamp = timestamp => {
 	const dateTime = momentFromTimestamp(timestamp);
-	return dateTime.utcOffset(0).format(config.excel.dateFormat);
+	return dateTime.utcOffset(0).format(DATE_FORMAT);
 };
 
 const timeFromTimestamp = timestamp => {
 	const dateTime = momentFromTimestamp(timestamp);
-	return dateTime.utcOffset(0).format(config.excel.timeFormat);
+	return dateTime.utcOffset(0).format(TIME_FORMAT);
 };
 
 module.exports = {
+	getToday,
 	getDaysInMilliseconds,
 	dateFromTimestamp,
 	timeFromTimestamp,

services/export/shared/requestAll.js

@@ -13,9 +13,8 @@
  * Removal or modification of this copyright notice is prohibited.
  *
  */
-const { Utils } = require('lisk-service-framework');
 
-const requestAllStandard = async (fn, params, limit) => {
+const requestAll = async (fn, params, limit) => {
 	const defaultMaxAmount = limit || 1000;
 	const oneRequestLimit = params.limit || 100;
 	const firstRequest = await fn({
@@ -48,57 +47,4 @@ const requestAllStandard = async (fn, params, limit) => {
 	return data;
 };
 
-const requestAllCustom = async (fn, method, params, limit) => {
-	const maxAmount = limit || Number.MAX_SAFE_INTEGER;
-	const oneRequestLimit = params.limit || 100;
-	const firstRequest = await fn(method, {
-		...params,
-		...{
-			limit: oneRequestLimit,
-			offset: 0,
-		},
-	});
-	const totalResponse = firstRequest;
-	if (totalResponse && !totalResponse.error) {
-		if (maxAmount > oneRequestLimit) {
-			for (let page = 1; page < Math.ceil(maxAmount / oneRequestLimit); page++) {
-				const curOffset = oneRequestLimit * page;
-
-				const result = await fn(method, {
-					...params,
-					...{
-						limit: Math.min(oneRequestLimit, maxAmount - curOffset),
-						offset: curOffset,
-					},
-				});
-
-				// This check needs to be updated for dynamic exit based on emptiness of object properties
-				if (!result || Utils.isEmptyArray(result) || Utils.isEmptyObject(result)) {
-					break;
-				}
-
-				if (Array.isArray(totalResponse)) totalResponse.push(...result);
-				else if (Utils.isObject(totalResponse)) {
-					// When response is an object, we should traverse the properties and merge the values.
-					// We can safely assume that the properties would be of type array, so concatenation will
-					// result in the whole response. If property is not an array, the latest value is kept.
-					Object.entries(totalResponse).forEach(([dataKey, dataVal]) => {
-						if (Array.isArray(dataVal)) {
-							totalResponse[dataKey].push(...result[dataKey]);
-						} else if (Utils.isObject(dataVal)) {
-							totalResponse[dataKey] = { ...totalResponse[dataKey], ...result[dataKey] };
-						} else {
-							totalResponse[dataKey] = result[dataKey];
-						}
-					});
-				}
-			}
-		}
-	}
-	return totalResponse;
-};
-
-module.exports = {
-	requestAllStandard,
-	requestAllCustom,
-};
+module.exports = requestAll;