fix: make indexing transactional

Signed-off-by: chaitika_ <[email protected]>

e2e/indexer.e2e-spec.ts

@@ -14,7 +14,7 @@ describe('Indexer', () => {
     beforeAll(async () => {
         const walletHelper = new WalletHelper();
         const bitcoinRPCUtil = new BitcoinRPCUtil();
-        const indexerService = new IndexerService({} as any);
+        const indexerService = new IndexerService();
         apiHelper = new ApiHelper();
 
         await bitcoinRPCUtil.createWallet('test_wallet');

jest.config.ts

@@ -17,6 +17,8 @@ const config: Config.InitialOptions = {
         '.*.entity.ts',
         '.*.dto.ts',
         '.*.spec.ts',
+        '.*.mock.ts',
+        '.*.fixture.ts',
         '.*.module-definition.ts',
         '.*.configuration.ts',
         '.*.configuration.model.ts',

src/block-data-providers/base-block-data-provider.abstract.ts

@@ -8,6 +8,8 @@ import {
 import { ConfigService } from '@nestjs/config';
 import { BlockStateService } from '@/block-state/block-state.service';
 import { BlockState } from '@/block-state/block-state.entity';
+import { EntityManager } from 'typeorm';
+import { OperationState } from '@/operation-state/operation-state.entity';
 
 export abstract class BaseBlockDataProvider<OperationState> {
     protected abstract readonly logger: Logger;
@@ -26,13 +28,15 @@ export abstract class BaseBlockDataProvider<OperationState> {
         vout: TransactionOutput[],
         blockHeight: number,
         blockHash: string,
+        manager: EntityManager,
     ): Promise<void> {
         await this.indexerService.index(
             txid,
             vin,
             vout,
             blockHeight,
             blockHash,
+            manager,
         );
     }
 
@@ -47,13 +51,14 @@ export abstract class BaseBlockDataProvider<OperationState> {
     async setState(
         state: OperationState,
         blockState: BlockState,
+        manager: EntityManager,
     ): Promise<void> {
-        await this.operationStateService.setOperationState(
-            this.operationStateKey,
-            state,
-        );
+        const operationState = new OperationState();
+        operationState.id = this.operationStateKey;
+        operationState.state = state;
 
-        await this.blockStateService.addBlockState(blockState);
+        await manager.save(OperationState, operationState);
+        await manager.save(BlockState, blockState);
     }
 
     abstract getBlockHash(height: number): Promise<string>;

src/block-data-providers/bitcoin-core/provider.spec.ts

@@ -13,6 +13,7 @@ import {
 } from '@/block-data-providers/bitcoin-core/provider-fixtures';
 import { Test, TestingModule } from '@nestjs/testing';
 import { BlockStateService } from '@/block-state/block-state.service';
+import { DbTransactionService } from '@/db-transaction/db-transaction.service';
 
 describe('Bitcoin Core Provider', () => {
     let provider: BitcoinCoreProvider;
@@ -51,6 +52,12 @@ describe('Bitcoin Core Provider', () => {
                     provide: BlockStateService,
                     useClass: jest.fn(),
                 },
+                {
+                    provide: DbTransactionService,
+                    useValue: {
+                        execute: jest.fn(),
+                    },
+                },
             ],
         }).compile();
 

src/block-data-providers/bitcoin-core/provider.ts

@@ -29,6 +29,7 @@ import { AxiosRequestConfig } from 'axios';
 import * as currency from 'currency.js';
 import { AxiosRetryConfig, makeRequest } from '@/common/request';
 import { BlockStateService } from '@/block-state/block-state.service';
+import { DbTransactionService } from '@/db-transaction/db-transaction.service';
 
 @Injectable()
 export class BitcoinCoreProvider
@@ -46,6 +47,7 @@ export class BitcoinCoreProvider
         indexerService: IndexerService,
         operationStateService: OperationStateService,
         blockStateService: BlockStateService,
+        private readonly dbTransactionService: DbTransactionService,
     ) {
         super(
             configService,
@@ -81,15 +83,18 @@ export class BitcoinCoreProvider
                     : 0;
             const blockHash = await this.getBlockHash(blockHeight);
 
-            await this.setState(
-                {
-                    indexedBlockHeight: blockHeight,
-                },
-                {
-                    blockHash,
-                    blockHeight,
-                },
-            );
+            await this.dbTransactionService.execute(async (manager) => {
+                await this.setState(
+                    {
+                        indexedBlockHeight: blockHeight,
+                    },
+                    {
+                        blockHash,
+                        blockHeight,
+                    },
+                    manager,
+                );
+            });
         }
     }
 
@@ -127,22 +132,29 @@ export class BitcoinCoreProvider
                     verbosityLevel,
                 );
 
-                for (const transaction of transactions) {
-                    const { txid, vin, vout, blockHeight, blockHash } =
-                        transaction;
-                    await this.indexTransaction(
-                        txid,
-                        vin,
-                        vout,
-                        blockHeight,
-                        blockHash,
+                await this.dbTransactionService.execute(async (manager) => {
+                    for (const transaction of transactions) {
+                        const { txid, vin, vout, blockHeight, blockHash } =
+                            transaction;
+                        await this.indexTransaction(
+                            txid,
+                            vin,
+                            vout,
+                            blockHeight,
+                            blockHash,
+                            manager,
+                        );
+                    }
+
+                    state.indexedBlockHeight = height;
+                    await this.setState(
+                        state,
+                        {
+                            blockHash: blockHash,
+                            blockHeight: height,
+                        },
+                        manager,
                     );
-                }
-
-                state.indexedBlockHeight = height;
-                await this.setState(state, {
-                    blockHash: blockHash,
-                    blockHeight: height,
                 });
             }
         } finally {

src/block-data-providers/block-provider.module.ts

@@ -9,13 +9,16 @@ import { BitcoinCoreProvider } from '@/block-data-providers/bitcoin-core/provide
 import { EsploraProvider } from '@/block-data-providers/esplora/provider';
 import { BlockStateService } from '@/block-state/block-state.service';
 import { BlockStateModule } from '@/block-state/block-state.module';
+import { DbTransactionModule } from '@/db-transaction/db-transaction.module';
+import { DbTransactionService } from '@/db-transaction/db-transaction.service';
 
 @Module({
     imports: [
         OperationStateModule,
         IndexerModule,
         ConfigModule,
         BlockStateModule,
+        DbTransactionModule,
     ],
     controllers: [],
     providers: [
@@ -26,12 +29,14 @@ import { BlockStateModule } from '@/block-state/block-state.module';
                 IndexerService,
                 OperationStateService,
                 BlockStateService,
+                DbTransactionService,
             ],
             useFactory: (
                 configService: ConfigService,
                 indexerService: IndexerService,
                 operationStateService: OperationStateService,
                 blockStateService: BlockStateService,
+                dbTransactionService: DbTransactionService,
             ) => {
                 switch (configService.get<ProviderType>('providerType')) {
                     case ProviderType.ESPLORA:
@@ -40,13 +45,15 @@ import { BlockStateModule } from '@/block-state/block-state.module';
                             indexerService,
                             operationStateService,
                             blockStateService,
+                            dbTransactionService,
                         );
                     case ProviderType.BITCOIN_CORE_RPC:
                         return new BitcoinCoreProvider(
                             configService,
                             indexerService,
                             operationStateService,
                             blockStateService,
+                            dbTransactionService,
                         );
                     default:
                         throw Error('unrecognised provider type in config');

src/block-data-providers/esplora/provider.ts

@@ -13,6 +13,7 @@ import {
 import { TAPROOT_ACTIVATION_HEIGHT } from '@/common/constants';
 import { BlockStateService } from '@/block-state/block-state.service';
 import { Cron, CronExpression } from '@nestjs/schedule';
+import { DbTransactionService } from '@/db-transaction/db-transaction.service';
 
 @Injectable()
 export class EsploraProvider
@@ -31,6 +32,7 @@ export class EsploraProvider
         indexerService: IndexerService,
         operationStateService: OperationStateService,
         blockStateService: BlockStateService,
+        private readonly dbTransactionService: DbTransactionService,
     ) {
         super(
             configService,
@@ -79,17 +81,20 @@ export class EsploraProvider
                     : 0;
             const blockHash = await this.getBlockHash(blockHeight);
 
-            await this.setState(
-                {
-                    currentBlockHeight: 0,
-                    indexedBlockHeight: blockHeight,
-                    lastProcessedTxIndex: 0, // we don't take coinbase txn into account
-                },
-                {
-                    blockHash,
-                    blockHeight,
-                },
-            );
+            await this.dbTransactionService.execute(async (manager) => {
+                await this.setState(
+                    {
+                        currentBlockHeight: 0,
+                        indexedBlockHeight: blockHeight,
+                        lastProcessedTxIndex: 0, // we don't take coinbase txn into account
+                    },
+                    {
+                        blockHash,
+                        blockHeight,
+                    },
+                    manager,
+                );
+            });
         }
     }
 
@@ -144,36 +149,45 @@ export class EsploraProvider
             );
 
             try {
-                await Promise.all(
-                    batch.map(async (txid) => {
-                        const tx = await this.getTx(txid);
-                        const vin: TransactionInput[] = tx.vin.map((input) => ({
-                            txid: input.txid,
-                            vout: input.vout,
-                            scriptSig: input.scriptsig,
-                            prevOutScript: input.prevout.scriptpubkey,
-                            witness: input.witness,
-                        }));
-                        const vout = tx.vout.map((output) => ({
-                            scriptPubKey: output.scriptpubkey,
-                            value: output.value,
-                        }));
-
-                        await this.indexTransaction(
-                            txid,
-                            vin,
-                            vout,
-                            height,
-                            hash,
-                        );
-                    }, this),
-                );
+                await this.dbTransactionService.execute(async (manager) => {
+                    await Promise.all(
+                        batch.map(async (txid) => {
+                            const tx = await this.getTx(txid);
+                            const vin: TransactionInput[] = tx.vin.map(
+                                (input) => ({
+                                    txid: input.txid,
+                                    vout: input.vout,
+                                    scriptSig: input.scriptsig,
+                                    prevOutScript: input.prevout.scriptpubkey,
+                                    witness: input.witness,
+                                }),
+                            );
+                            const vout = tx.vout.map((output) => ({
+                                scriptPubKey: output.scriptpubkey,
+                                value: output.value,
+                            }));
+
+                            await this.indexTransaction(
+                                txid,
+                                vin,
+                                vout,
+                                height,
+                                hash,
+                                manager,
+                            );
+                        }, this),
+                    );
 
-                state.indexedBlockHeight = height;
-                state.lastProcessedTxIndex = i + this.batchSize - 1;
-                await this.setState(state, {
-                    blockHeight: height,
-                    blockHash: hash,
+                    state.indexedBlockHeight = height;
+                    state.lastProcessedTxIndex = i + this.batchSize - 1;
+                    await this.setState(
+                        state,
+                        {
+                            blockHeight: height,
+                            blockHash: hash,
+                        },
+                        manager,
+                    );
                 });
             } catch (error) {
                 this.logger.error(

src/block-state/block-state.service.ts

@@ -23,10 +23,6 @@ export class BlockStateService {
         )[0];
     }
 
-    async addBlockState(state: BlockState): Promise<void> {
-        await this.blockStateRepository.save(state);
-    }
-
     async removeState(state: BlockState): Promise<void> {
         await this.blockStateRepository.delete(state.blockHeight);
         await this.transactionService.deleteTransactionByBlockHash(

src/db-transaction/db-transaction.mock.ts

@@ -0,0 +1,43 @@
+import 'jest';
+
+export const mockQueryBuilder = {
+    select: jest.fn().mockReturnThis(),
+    addSelect: jest.fn().mockReturnThis(),
+    where: jest.fn().mockReturnThis(),
+    groupBy: jest.fn().mockReturnThis(),
+    createQueryBuilder: jest.fn().mockReturnThis(),
+    leftJoin: jest.fn().mockReturnThis(),
+    leftJoinAndSelect: jest.fn().mockReturnThis(),
+    andWhere: jest.fn().mockReturnThis(),
+    getOne: jest.fn(),
+    getQuery: jest.fn(),
+    getRawOne: jest.fn(),
+    from: jest.fn().mockReturnThis(),
+};
+
+export const mockEntityManager = {
+    createQueryBuilder: jest.fn().mockReturnValue(mockQueryBuilder),
+    query: jest.fn(),
+    save: jest.fn(),
+    update: jest.fn(),
+    upsert: jest.fn(),
+    find: jest.fn(),
+    findOne: jest.fn(),
+    findOneBy: jest.fn(),
+    increment: jest.fn(),
+    findBy: jest.fn(),
+    exists: jest.fn(),
+};
+
+export const queryRunnerMock = {
+    connect: jest.fn(),
+    startTransaction: jest.fn(),
+    commitTransaction: jest.fn(),
+    rollbackTransaction: jest.fn(),
+    release: jest.fn(),
+    manager: mockEntityManager,
+};
+
+export class MockDataSource {
+    createQueryRunner = jest.fn().mockReturnValue(queryRunnerMock);
+}