refactor(core-transactions): bootstrap transactions in batches

__tests__/unit/crypto/blocks/factory.test.ts

@@ -15,7 +15,7 @@ export const expectBlock = ({ data }: { data: IBlockData }) => {
     delete blockWithoutTransactions.transactions;
 
     expect(data).toEqual(blockWithoutTransactions);
-}
+};
 
 beforeEach(() => configManager.setFromPreset("devnet"));
 
@@ -40,7 +40,7 @@ describe("BlockFactory", () => {
         it("should create a block with exceptions", () => {
             // @ts-ignore
             expect(() => BlockFactory.fromData(blockWithExceptions)).not.toThrow();
-        })
+        });
     });
 
     describe(".fromJson", () => {

__tests__/unit/crypto/fixtures/block.ts

@@ -158,50 +158,53 @@ export const dummyBlock3 = {
 };
 
 export const blockWithExceptions = {
-    "id": "15986760691378510176",
-    "version": 0,
-    "timestamp": 52310186,
-    "height": 766787,
-    "reward": "200000000",
-    "transactions": [
+    id: "15986760691378510176",
+    version: 0,
+    timestamp: 52310186,
+    height: 766787,
+    reward: "200000000",
+    transactions: [
         {
-            "version": 1,
-            "network": 30,
-            "type": 0,
-            "timestamp": 52308692,
-            "senderPublicKey": "039c09c66328ebe7569cd49aa441e45c9df96c1d644c7acf33f7bc763709c41c1f",
-            "fee": "10000000",
-            "amount": "0",
-            "vendorFieldHex": "54727565205061737461205765696768742e",
-            "expiration": 0,
-            "recipientId": "D5KU9KrMYXdkEsRbv4y8hvetGbsJwf9z3P",
-            "signature": "304402200ca79a6ebcd29cbf6bea42fa6890e0c7bfd82fd5e7cd327913f966b433b24086022051f127b1349af1ccb49cdc27595844d5228d612ef8c7d57d63dc5b75620c7e64",
-            "vendorField": "True Pasta Weight.",
-            "id": "3945e67bb5e864d2dd206293f1e778fa2181db5f81c2efc0a89e8fe53e2a2e7c"
+            version: 1,
+            network: 30,
+            type: 0,
+            timestamp: 52308692,
+            senderPublicKey: "039c09c66328ebe7569cd49aa441e45c9df96c1d644c7acf33f7bc763709c41c1f",
+            fee: "10000000",
+            amount: "0",
+            vendorFieldHex: "54727565205061737461205765696768742e",
+            expiration: 0,
+            recipientId: "D5KU9KrMYXdkEsRbv4y8hvetGbsJwf9z3P",
+            signature:
+                "304402200ca79a6ebcd29cbf6bea42fa6890e0c7bfd82fd5e7cd327913f966b433b24086022051f127b1349af1ccb49cdc27595844d5228d612ef8c7d57d63dc5b75620c7e64",
+            vendorField: "True Pasta Weight.",
+            id: "3945e67bb5e864d2dd206293f1e778fa2181db5f81c2efc0a89e8fe53e2a2e7c",
         },
         {
-            "version": 1,
-            "network": 30,
-            "type": 0,
-            "timestamp": 52308443,
-            "senderPublicKey": "02364aaaf17c35f74e397433f988361aee408c5b5314eaad5b815c4f4f7c578b1e",
-            "fee": "10000000",
-            "amount": "250000000",
-            "vendorFieldHex": "54727565205061737461205765696768742e",
-            "expiration": 0,
-            "recipientId": "DJA2sqCbnmR63sD8doGrXrK3fCiqcA4GUw",
-            "signature": "3045022100f3a61a0460abe78a4705904216ac17a7bf50b19d70717624c15b64a94207cbeb022070a7bf3d97a3e690839ce1764413ce890ea233eb7eba202d373a376fc4b551b6",
-            "vendorField": "True Pasta Weight.",
-            "id": "8e4853ed764bb1853e58fda4a9c507f73b820dd738e4a9b197dafb37a22b6b0f"
-        }
+            version: 1,
+            network: 30,
+            type: 0,
+            timestamp: 52308443,
+            senderPublicKey: "02364aaaf17c35f74e397433f988361aee408c5b5314eaad5b815c4f4f7c578b1e",
+            fee: "10000000",
+            amount: "250000000",
+            vendorFieldHex: "54727565205061737461205765696768742e",
+            expiration: 0,
+            recipientId: "DJA2sqCbnmR63sD8doGrXrK3fCiqcA4GUw",
+            signature:
+                "3045022100f3a61a0460abe78a4705904216ac17a7bf50b19d70717624c15b64a94207cbeb022070a7bf3d97a3e690839ce1764413ce890ea233eb7eba202d373a376fc4b551b6",
+            vendorField: "True Pasta Weight.",
+            id: "8e4853ed764bb1853e58fda4a9c507f73b820dd738e4a9b197dafb37a22b6b0f",
+        },
     ],
-    "previousBlock": "17296711371070322453",
-    "numberOfTransactions": 2,
-    "totalAmount": "250000000",
-    "totalFee": "20000000",
-    "payloadLength": 64,
-    "payloadHash": "53033d24a4f133e47fee3871d82bca12bdf8b58ec4ebf43ac2b7849e78278256",
-    "generatorPublicKey": "02d0244d939fad9004cc104f71b46b428d903e4f2988a65f39fdaa1b7482894c9e",
-    "blockSignature": "30440220053747c5d02b38c9f8980d771918a5c48c99080a0bba661d3609553ed14c3f0d022063da2ba62e002d91b0f3bdba4e0c822bf0f5ce88b95867192252a2418ec0556f",
-    "previousBlockHex": "f00a43599f1b3715"
-}
+    previousBlock: "17296711371070322453",
+    numberOfTransactions: 2,
+    totalAmount: "250000000",
+    totalFee: "20000000",
+    payloadLength: 64,
+    payloadHash: "53033d24a4f133e47fee3871d82bca12bdf8b58ec4ebf43ac2b7849e78278256",
+    generatorPublicKey: "02d0244d939fad9004cc104f71b46b428d903e4f2988a65f39fdaa1b7482894c9e",
+    blockSignature:
+        "30440220053747c5d02b38c9f8980d771918a5c48c99080a0bba661d3609553ed14c3f0d022063da2ba62e002d91b0f3bdba4e0c822bf0f5ce88b95867192252a2418ec0556f",
+    previousBlockHex: "f00a43599f1b3715",
+};

packages/core-database-postgres/src/queries/index.ts

@@ -33,6 +33,7 @@ export const queries = {
         lastForgedBlocks: loadQueryFile(__dirname, "./state-builder/last-forged-blocks.sql"),
         receivedTransactions: loadQueryFile(__dirname, "./state-builder/received-transactions.sql"),
         sentTransactions: loadQueryFile(__dirname, "./state-builder/sent-transactions.sql"),
+        countType: loadQueryFile(__dirname, "./state-builder/count-type.sql"),
         assetsByType: loadQueryFile(__dirname, "./state-builder/assets-by-type.sql"),
     },
     transactions: {

packages/core-database-postgres/src/queries/state-builder/assets-by-type.sql

@@ -9,3 +9,4 @@ SELECT sender_public_key,
 FROM transactions
 WHERE type = ${type} AND type_group = ${typeGroup}
 ORDER BY timestamp ASC, sequence ASC
+LIMIT ${limit} OFFSET ${offset}

packages/core-database-postgres/src/queries/state-builder/count-type.sql

@@ -0,0 +1,3 @@
+SELECT COUNT(id) as "count"
+FROM transactions
+WHERE type = ${type} AND type_group = ${typeGroup}

packages/core-database-postgres/src/repositories/transactions.ts

@@ -165,8 +165,12 @@ export class TransactionsRepository extends Repository implements Database.ITran
         return this.db.manyOrNone(queries.transactions.latestByBlocks, { ids });
     }
 
-    public async getAssetsByType(type: number, typeGroup: number = Enums.TransactionTypeGroup.Core): Promise<any> {
-        return this.db.manyOrNone(queries.stateBuilder.assetsByType, { typeGroup, type });
+    public async getCountOfType(type: number, typeGroup: number = Enums.TransactionTypeGroup.Core): Promise<any> {
+        return +(await this.db.one(queries.stateBuilder.countType, { typeGroup, type })).count;
+    }
+
+    public async getAssetsByType(type: number, typeGroup: number, limit: number, offset: number): Promise<any> {
+        return this.db.manyOrNone(queries.stateBuilder.assetsByType, { typeGroup, type, limit, offset });
     }
 
     public async getReceivedTransactions(): Promise<any> {

packages/core-database/src/repositories/transactions-business-repository.ts

@@ -87,8 +87,17 @@ export class TransactionsBusinessRepository implements Database.ITransactionsBus
         );
     }
 
-    public async getAssetsByType(type: number, typeGroup?: number): Promise<any> {
-        return this.databaseServiceProvider().connection.transactionsRepository.getAssetsByType(type, typeGroup);
+    public async getCountOfType(type: number, typeGroup?: number): Promise<number> {
+        return this.databaseServiceProvider().connection.transactionsRepository.getCountOfType(type, typeGroup);
+    }
+
+    public async getAssetsByType(type: number, typeGroup: number, limit: number, offset: number): Promise<any> {
+        return this.databaseServiceProvider().connection.transactionsRepository.getAssetsByType(
+            type,
+            typeGroup,
+            limit,
+            offset,
+        );
     }
 
     public async getReceivedTransactions(): Promise<any> {

packages/core-interfaces/src/core-database/business-repository/transactions-business-repository.ts

@@ -23,7 +23,9 @@ export interface ITransactionsBusinessRepository {
 
     findByTypeAndId(type: number, id: string): Promise<Interfaces.ITransactionData>;
 
-    getAssetsByType(type: number, typeGroup?: number): Promise<any>;
+    getCountOfType(type: number, typeGroup?: number): Promise<number>;
+
+    getAssetsByType(type: number, typeGroup: number, limit: number, offset: number): Promise<any>;
 
     getReceivedTransactions(): Promise<any>;
 

packages/core-interfaces/src/core-database/database-repository/transactions-repository.ts

@@ -46,7 +46,9 @@ export interface ITransactionsRepository extends IRepository {
         }>
     >;
 
-    getAssetsByType(type: number, typeGroup?: number): Promise<IBootstrapTransaction[]>;
+    getCountOfType(type: number, typeGroup?: number): Promise<number>;
+
+    getAssetsByType(type: number, typeGroup: number, limit: number, offset: number): Promise<IBootstrapTransaction[]>;
 
     getReceivedTransactions(): Promise<any>;
 

packages/core-magistrate-transactions/src/handlers/bridgechain-registration.ts

@@ -1,6 +1,6 @@
 import { Database, EventEmitter, State, TransactionPool } from "@arkecosystem/core-interfaces";
 import { Transactions as MagistrateTransactions } from "@arkecosystem/core-magistrate-crypto";
-import { Handlers } from "@arkecosystem/core-transactions";
+import { Handlers, TransactionReader } from "@arkecosystem/core-transactions";
 import { Interfaces, Managers, Transactions, Utils } from "@arkecosystem/crypto";
 import { BusinessIsResignedError, WalletIsNotBusinessError } from "../errors";
 import { MagistrateApplicationEvents } from "../events";
@@ -26,27 +26,29 @@ export class BridgechainRegistrationTransactionHandler extends Handlers.Transact
     }
 
     public async bootstrap(connection: Database.IConnection, walletManager: State.IWalletManager): Promise<void> {
-        const transactions: Database.IBootstrapTransaction[] = await connection.transactionsRepository.getAssetsByType(
-            this.getConstructor().type,
-            this.getConstructor().typeGroup,
-        );
-        for (const transaction of transactions) {
-            const wallet: State.IWallet = walletManager.findByPublicKey(transaction.senderPublicKey);
-            const businessAttributes: IBusinessWalletAttributes = wallet.getAttribute<IBusinessWalletAttributes>(
-                "business",
-            );
-            if (!businessAttributes.bridgechains) {
-                businessAttributes.bridgechains = {};
+        const reader: TransactionReader = await TransactionReader.create(connection, this.getConstructor());
+
+        while (reader.hasNext()) {
+            const transactions = await reader.read();
+
+            for (const transaction of transactions) {
+                const wallet: State.IWallet = walletManager.findByPublicKey(transaction.senderPublicKey);
+                const businessAttributes: IBusinessWalletAttributes = wallet.getAttribute<IBusinessWalletAttributes>(
+                    "business",
+                );
+                if (!businessAttributes.bridgechains) {
+                    businessAttributes.bridgechains = {};
+                }
+
+                const bridgechainId: Utils.BigNumber = this.getBridgechainId(walletManager);
+                businessAttributes.bridgechains[bridgechainId.toFixed()] = {
+                    bridgechainId,
+                    bridgechainAsset: transaction.asset.bridgechainRegistration,
+                };
+
+                wallet.setAttribute<IBusinessWalletAttributes>("business", businessAttributes);
+                walletManager.reindex(wallet);
             }
-
-            const bridgechainId: Utils.BigNumber = this.getBridgechainId(walletManager);
-            businessAttributes.bridgechains[bridgechainId.toFixed()] = {
-                bridgechainId,
-                bridgechainAsset: transaction.asset.bridgechainRegistration,
-            };
-
-            wallet.setAttribute<IBusinessWalletAttributes>("business", businessAttributes);
-            walletManager.reindex(wallet);
         }
     }
 

packages/core-magistrate-transactions/src/handlers/bridgechain-resignation.ts

@@ -1,7 +1,7 @@
 import { Database, EventEmitter, State, TransactionPool } from "@arkecosystem/core-interfaces";
 import { Transactions as MagistrateTransactions } from "@arkecosystem/core-magistrate-crypto";
 import { Interfaces as MagistrateInterfaces } from "@arkecosystem/core-magistrate-crypto";
-import { Handlers } from "@arkecosystem/core-transactions";
+import { Handlers, TransactionReader } from "@arkecosystem/core-transactions";
 import { Interfaces, Managers, Transactions } from "@arkecosystem/crypto";
 import {
     BridgechainIsNotRegisteredError,
@@ -31,22 +31,24 @@ export class BridgechainResignationTransactionHandler extends Handlers.Transacti
     }
 
     public async bootstrap(connection: Database.IConnection, walletManager: State.IWalletManager): Promise<void> {
-        const transactions: Database.IBootstrapTransaction[] = await connection.transactionsRepository.getAssetsByType(
-            this.getConstructor().type,
-            this.getConstructor().typeGroup,
-        );
-        for (const transaction of transactions) {
-            const wallet: State.IWallet = walletManager.findByPublicKey(transaction.senderPublicKey);
-            const businessAttributes: IBusinessWalletAttributes = wallet.getAttribute<IBusinessWalletAttributes>(
-                "business",
-            );
+        const reader: TransactionReader = await TransactionReader.create(connection, this.getConstructor());
+
+        while (reader.hasNext()) {
+            const transactions = await reader.read();
+
+            for (const transaction of transactions) {
+                const wallet: State.IWallet = walletManager.findByPublicKey(transaction.senderPublicKey);
+                const businessAttributes: IBusinessWalletAttributes = wallet.getAttribute<IBusinessWalletAttributes>(
+                    "business",
+                );
 
-            const bridgechainAsset =
-                businessAttributes.bridgechains[transaction.asset.bridgechainResignation.bridgechainId];
-            bridgechainAsset.resigned = true;
+                const bridgechainAsset =
+                    businessAttributes.bridgechains[transaction.asset.bridgechainResignation.bridgechainId];
+                bridgechainAsset.resigned = true;
 
-            wallet.setAttribute<IBusinessWalletAttributes>("business", businessAttributes);
-            walletManager.reindex(wallet);
+                wallet.setAttribute<IBusinessWalletAttributes>("business", businessAttributes);
+                walletManager.reindex(wallet);
+            }
         }
     }
 

packages/core-magistrate-transactions/src/handlers/bridgechain-update.ts

@@ -5,7 +5,7 @@ import {
     Interfaces as MagistrateInterfaces,
     Transactions as MagistrateTransactions,
 } from "@arkecosystem/core-magistrate-crypto";
-import { Handlers } from "@arkecosystem/core-transactions";
+import { Handlers, TransactionReader } from "@arkecosystem/core-transactions";
 import { Interfaces, Managers, Transactions } from "@arkecosystem/crypto";
 import {
     BridgechainIsNotRegisteredError,
@@ -35,20 +35,22 @@ export class BridgechainUpdateTransactionHandler extends Handlers.TransactionHan
     }
 
     public async bootstrap(connection: Database.IConnection, walletManager: State.IWalletManager): Promise<void> {
-        const transactions: Database.IBootstrapTransaction[] = await connection.transactionsRepository.getAssetsByType(
-            this.getConstructor().type,
-            this.getConstructor().typeGroup,
-        );
-        for (const transaction of transactions) {
-            const wallet: State.IWallet = walletManager.findByPublicKey(transaction.senderPublicKey);
-            const businessAttributes: IBusinessWalletAttributes = wallet.getAttribute<IBusinessWalletAttributes>(
-                "business",
-            );
+        const reader: TransactionReader = await TransactionReader.create(connection, this.getConstructor());
+
+        while (reader.hasNext()) {
+            const transactions = await reader.read();
+
+            for (const transaction of transactions) {
+                const wallet: State.IWallet = walletManager.findByPublicKey(transaction.senderPublicKey);
+                const businessAttributes: IBusinessWalletAttributes = wallet.getAttribute<IBusinessWalletAttributes>(
+                    "business",
+                );
 
-            const { bridgechainId, seedNodes } = transaction.asset.bridgechainUpdate;
-            businessAttributes.bridgechains[bridgechainId].bridgechainAsset.seedNodes = seedNodes;
+                const { bridgechainId, seedNodes } = transaction.asset.bridgechainUpdate;
+                businessAttributes.bridgechains[bridgechainId].bridgechainAsset.seedNodes = seedNodes;
 
-            walletManager.reindex(wallet);
+                walletManager.reindex(wallet);
+            }
         }
     }
 
@@ -127,13 +129,12 @@ export class BridgechainUpdateTransactionHandler extends Handlers.TransactionHan
             "business",
         );
 
-        const transactionsRepository: Database.ITransactionsRepository = app.resolvePlugin<Database.IConnection>(
-            "database",
-        ).transactionsRepository;
-        const updateTransactions: Database.IBootstrapTransaction[] = await transactionsRepository.getAssetsByType(
-            Enums.MagistrateTransactionType.BridgechainUpdate,
-            Enums.MagistrateTransactionGroup,
-        );
+        const connection: Database.IConnection = app.resolvePlugin<Database.IConnection>("database");
+        const reader: TransactionReader = await TransactionReader.create(connection, this.getConstructor());
+        const updateTransactions: Database.IBootstrapTransaction[] = [];
+        while (reader.hasNext()) {
+            updateTransactions.push(...(await reader.read()));
+        }
 
         if (updateTransactions.length > 1) {
             const updateTransaction: Database.IBootstrapTransaction = updateTransactions.pop();