Merge pull request #4 from dappforce/deploy/datasource-subquery

feat: bumped build

deployment/feature-based/all.yaml

@@ -59,20 +59,22 @@ data:
   AGGREGATOR_REDIS_PREFIX: 'aggregator_queue_<BRANCH>'
   AGGREGATOR_REDIS_PORT: '6379'
   AGGREGATOR_REDIS_ENABLE_SSL: 'false'
-  AGGREGATOR_HISTORY_RENEW_INTERVAL_MS: '60000'
-  AGGREGATOR_GS_MAIN_CHUNK_BLOCKS_SIZE: '1000000'
+  AGGREGATOR_HISTORY_RENEW_INTERVAL_MS: '120000'
+  AGGREGATOR_GS_MAIN_CHUNK_BLOCKS_SIZE: '2000000'
 
   DATA_SOURCE__SUBSQUID__POLKADOT__TRANSFER: 'https://squid.subsquid.io/gs-main-polkadot/graphql'
   DATA_SOURCE__SUBSQUID__KUSAMA__TRANSFER: 'https://squid.subsquid.io/gs-main-kusama/graphql'
   DATA_SOURCE__SUBSQUID__MOONBEAM__TRANSFER: 'https://squid.subsquid.io/gs-main-moonbeam/graphql'
   DATA_SOURCE__SUBSQUID__MOONRIVER__TRANSFER: 'https://squid.subsquid.io/gs-main-moonriver/graphql'
   DATA_SOURCE__SUBSQUID__ASTAR__TRANSFER: 'https://squid.subsquid.io/gs-main-astar/graphql'
+  DATA_SOURCE__SUBSQUID__SUBSOCIAL__TRANSFER: 'https://squid.subsquid.io/gs-main-astar/graphql'
 
   DATA_SOURCE__SUBQUERY__POLKADOT__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-polkadot'
   DATA_SOURCE__SUBQUERY__KUSAMA__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-kusama'
   DATA_SOURCE__SUBQUERY__MOONBEAM__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-moonbeam'
   DATA_SOURCE__SUBQUERY__MOONRIVER__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-moonriver'
   DATA_SOURCE__SUBQUERY__ASTAR__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-astar'
+  DATA_SOURCE__SUBQUERY__SUBSOCIAL__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-subsocial'
 
   DATA_SOURCE_PROVIDER_TRANSFER: 'SUBQUERY'
 

deployment/prod/all.yaml

@@ -73,6 +73,7 @@ data:
   DATA_SOURCE__SUBQUERY__MOONBEAM__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-moonbeam'
   DATA_SOURCE__SUBQUERY__MOONRIVER__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-moonriver'
   DATA_SOURCE__SUBQUERY__ASTAR__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-astar'
+  DATA_SOURCE__SUBQUERY__SUBSOCIAL__TRANSFER: 'https://api.subquery.network/sq/nova-wallet/nova-wallet-subsocial'
 
   DATA_SOURCE_PROVIDER_TRANSFER: 'SUBQUERY'
 

package.json

@@ -24,6 +24,7 @@
     "codegen:subquery-nova-query": "graphql-codegen --require dotenv/config --config codegenConfig/codegen-subquery-nova-query.ts"
   },
   "dependencies": {
+    "@apollo/client": "^3.8.8",
     "@apollo/server": "^4.7.2",
     "@bull-board/api": "^5.8.4",
     "@bull-board/express": "^5.8.4",
@@ -38,6 +39,7 @@
     "@nestjs/schedule": "^4.0.0",
     "@nestjs/typeorm": "^9.0.1",
     "@polkadot/util-crypto": "^12.4.2",
+    "@subql/apollo-links": "^1.2.5",
     "@subsocial/api": "^0.8.14",
     "@subsocial/utils": "^0.8.14",
     "@subsquid/openreader": "^3.1.7",

schema.gql

@@ -20,6 +20,7 @@ enum BlockchainTag {
   MOONBEAM
   MOONRIVER
   ASTAR
+  SUBSOCIAL
 }
 
 type TransferNative {
@@ -133,6 +134,7 @@ input FindAccountTxHistoryWhereParams {
   publicKey: String
   blockchainTag: [BlockchainTag!]
   txKind: [TransactionKind!]
+  success: Boolean = true
 }
 
 enum FindAccountTxHistoryOrderBy {

src/app.module.ts

@@ -53,6 +53,7 @@ dotenv.config();
     AccountSyncSchedulerModule
   ],
 })
+
 export class AppModule implements NestModule {
   configure(consumer: MiddlewareConsumer) {
     // consumer.apply(ApiToggleMiddleware).forRoutes('*');

src/config.module.ts

@@ -90,6 +90,9 @@ export class AppConfig {
 
   @IsNotEmpty()
   readonly DATA_SOURCE__SUBQUERY__ASTAR__TRANSFER: string;
+
+  @IsNotEmpty()
+  readonly DATA_SOURCE__SUBQUERY__SUBSOCIAL__TRANSFER: string;
 }
 
 @Global()

src/constants/blockchain.ts

@@ -7,8 +7,8 @@ export enum BlockchainTag {
   MOONBEAM = 'MOONBEAM',
   MOONRIVER = 'MOONRIVER',
   ASTAR = 'ASTAR',
+  SUBSOCIAL = 'SUBSOCIAL',
 }
-
 // https://github.com/polkadot-js/apps/blob/7c12692ee34aadd815b282ff4dca3d0a4763a9ca/packages/apps-config/src/endpoints/productionRelayPolkadot.ts#L645
 
 // https://github.com/paritytech/ss58-registry/blob/main/ss58-registry.json
@@ -64,39 +64,56 @@ export const supportedBlockchainDetails: Omit<Blockchain, 'id'>[] = [
     logo: '',
     color: '#1b6dc1d9',
   },
+  {
+    prefix: 28,
+    text: 'Subsocial',
+    info: 'subsocial',
+    tag: BlockchainTag.SUBSOCIAL,
+    decimal: 10,
+    symbols: ['SUB'],
+    logo: '',
+    color: '#1b6dc1d9',
+  },
 ];
 
 export const blockchainDataSourceConfigs = [
+  // {
+  //   tag: BlockchainTag.POLKADOT,
+  //   events: {
+  //     [NativeTransactionKind.TRANSFER]: null,
+  //     // [NativeTransactionKind.REWARD]: null,
+  //   },
+  // },
+  // {
+  //   tag: BlockchainTag.KUSAMA,
+  //   events: {
+  //     [NativeTransactionKind.TRANSFER]: null,
+  //     // [NativeTransactionKind.REWARD]: null,
+  //   },
+  // },
+  // {
+  //   tag: BlockchainTag.MOONBEAM,
+  //   events: {
+  //     [NativeTransactionKind.TRANSFER]: null,
+  //     // [NativeTransactionKind.REWARD]: null,
+  //   },
+  // },
+  // {
+  //   tag: BlockchainTag.MOONRIVER,
+  //   events: {
+  //     [NativeTransactionKind.TRANSFER]: null,
+  //     // [NativeTransactionKind.REWARD]: null,
+  //   },
+  // },
+  // {
+  //   tag: BlockchainTag.ASTAR,
+  //   events: {
+  //     [NativeTransactionKind.TRANSFER]: null,
+  //     // [NativeTransactionKind.REWARD]: null,
+  //   },
+  // },
   {
-    tag: BlockchainTag.POLKADOT,
-    events: {
-      [NativeTransactionKind.TRANSFER]: null,
-      // [NativeTransactionKind.REWARD]: null,
-    },
-  },
-  {
-    tag: BlockchainTag.KUSAMA,
-    events: {
-      [NativeTransactionKind.TRANSFER]: null,
-      // [NativeTransactionKind.REWARD]: null,
-    },
-  },
-  {
-    tag: BlockchainTag.MOONBEAM,
-    events: {
-      [NativeTransactionKind.TRANSFER]: null,
-      // [NativeTransactionKind.REWARD]: null,
-    },
-  },
-  {
-    tag: BlockchainTag.MOONRIVER,
-    events: {
-      [NativeTransactionKind.TRANSFER]: null,
-      // [NativeTransactionKind.REWARD]: null,
-    },
-  },
-  {
-    tag: BlockchainTag.ASTAR,
+    tag: BlockchainTag.SUBSOCIAL,
     events: {
       [NativeTransactionKind.TRANSFER]: null,
       // [NativeTransactionKind.REWARD]: null,

src/dependencyServiceModule.module.ts

@@ -1,8 +1,9 @@
-import { Module } from '@nestjs/common';
+import { Global, Module } from '@nestjs/common';
 import { CryptoUtils } from './utils/cryptoUtils';
 import { CommonUtils } from './utils/commonUtils';
 import { DataSourceUtils } from './utils/dataSources/dataSourceUtils';
 
+@Global()
 @Module({
   providers: [CryptoUtils, CommonUtils, DataSourceUtils],
   exports: [CryptoUtils, CommonUtils, DataSourceUtils],

src/modules/accountSyncScheduler/accountSyncScheduler.service.ts

@@ -56,7 +56,7 @@ export class AccountSyncSchedulerService {
   }: {
     publicKey: string;
   }) {
-    const decoratedPublicKey = this.cryptoUtils.addressToHex(publicKey);
+    const decoratedPublicKey = this.cryptoUtils.addressToHexIfNotHex(publicKey);
     const subscriptionData = await this.getOrCreateHistoryUpdateSubEntity(
       decoratedPublicKey,
     );

src/modules/apiGateway/gql/dto/input/findTransactionsHistory.input.dto.ts

@@ -18,6 +18,9 @@ export class FindAccountTxHistoryWhereParams {
 
   @Field(() => [TransactionKind!], { nullable: true })
   txKind?: TransactionKind[];
+
+  @Field(() => Boolean, { nullable: true, defaultValue: true })
+  success?: boolean;
 }
 
 @InputType()

src/modules/dataAggregator/dto/dataSourceEmptyResponsePlaceholder.dto.ts

@@ -0,0 +1,3 @@
+export class DataSourceEmptyResponsePlaceholderDto {
+  resultPlaceholder?: Record<any, any>
+}

src/modules/dataAggregator/dto/getIndexerLastProcessedHeight.args.dto.ts

@@ -1,3 +1,6 @@
-export class GetIndexerLastProcessedHeightArgs {
-  queryUrl: string;
+import { ApolloLink } from '@apollo/client/core';
+import { DataSourceEmptyResponsePlaceholderDto } from './dataSourceEmptyResponsePlaceholder.dto';
+
+export class GetIndexerLastProcessedHeightArgs extends DataSourceEmptyResponsePlaceholderDto {
+  queryUrl: string | ApolloLink;
 }

src/modules/dataAggregator/dto/getTransfersByAccount.args.dto.ts

@@ -1,12 +1,14 @@
 import { BlockchainTag } from '../../../constants/blockchain';
+import { ApolloLink } from '@apollo/client/core';
+import { DataSourceEmptyResponsePlaceholderDto } from './dataSourceEmptyResponsePlaceholder.dto';
 
-export class GetTransfersByAccountArgs {
+export class GetTransfersByAccountArgs extends DataSourceEmptyResponsePlaceholderDto {
   blockchainTag: BlockchainTag;
   limit: number;
   offset: number;
   blockNumber_gt: number;
   blockNumber_lt: number | null;
   publicKey?: string;
   address?: string;
-  queryUrl: string;
+  queryUrl: string | ApolloLink;
 }

src/modules/dataAggregator/dto/transfersByAccountDecorated.dto.ts

@@ -1,6 +1,7 @@
 import { TransferDirection } from '../../../constants/common';
 
 export type TransferDecoratedDto = {
+  id: string;
   direction?: TransferDirection | null;
   transfer?: {
     id: string;

src/modules/dataAggregator/services/aggregation.helper.ts

@@ -86,6 +86,10 @@ export class AggregationHelper {
         queryUrl: inputData.sourceUrl,
       });
 
+    console.log(
+      `${inputData.blockchainTag} indexer height - ${sourceIndexerStatus.height}`,
+    );
+
     let chunksRanges = this.getChunksRanges({
       latestProcessedBlock: inputData.latestProcessedBlock,
       totalBlocks: sourceIndexerStatus.height,
@@ -251,7 +255,11 @@ export class AggregationHelper {
     const pageSize = 1000;
     await runQuery();
     return {
-      fetchedChunkData: responseBuffer,
+      fetchedChunkData:
+        this.dataSourceUtils.getListWithoutDuplicates<TransferDecoratedDto>(
+          responseBuffer,
+          'id',
+        ),
     };
   }
 

src/modules/dataAggregator/services/dataAggregator.service.ts

@@ -24,7 +24,7 @@ export class DataAggregatorService {
     const data: RefreshAccountTxHistoryJobDataDto = job.data;
     // TODO add management of top level request from client
 
-    const publicKeyDecorated = this.cryptoUtils.addressToHex(data.publicKey);
+    const publicKeyDecorated = this.cryptoUtils.addressToHexIfNotHex(data.publicKey);
     const txAccount = await this.accountService.getOrCreateAccount(
       publicKeyDecorated,
     );

src/modules/entities/accountTransaction/accountTransaction.service.ts

@@ -44,15 +44,16 @@ export class AccountTransactionService {
   }
 
   async findAccountTxHistory({
-    where: { publicKey, blockchainTag, txKind },
+    where: { publicKey, blockchainTag, txKind, success },
     orderDirection,
     orderBy,
     pageSize,
     offset,
   }: FindAccountTxHistoryArgs): Promise<[AccountTransaction[], number]> {
     return this.accountTransactionRepository.findAndCount({
       where: {
-        ownerPublicKey: this.cryptoUtils.addressToHex(publicKey),
+        ownerPublicKey: this.cryptoUtils.addressToHexIfNotHex(publicKey),
+        success,
         ...(blockchainTag ? { blockchainTag: In(blockchainTag) } : {}),
         ...(txKind ? { txKind: In(txKind) } : {}),
       },

src/modules/entities/blockchain/blockchain.service.ts

@@ -8,6 +8,8 @@ import {
   supportedBlockchainDetails,
 } from '../../../constants/blockchain';
 import { AppConfig } from '../../../config.module';
+import { DataSourceUtils } from '../../../utils/dataSources/dataSourceUtils';
+import { NativeTransactionKind } from '../../../constants/common';
 
 @Injectable()
 export class BlockchainService {
@@ -17,6 +19,7 @@ export class BlockchainService {
     @InjectRepository(Blockchain)
     public readonly blockchainRepository: Repository<Blockchain>,
     private appConfig: AppConfig,
+    private dataSourceUtils: DataSourceUtils,
   ) {
     this.setDataSourceEndpoints();
   }
@@ -27,14 +30,15 @@ export class BlockchainService {
 
       for (const eventName in chainConfig.events) {
         chainConfigUpdated.events[eventName] =
-          this.appConfig[
-            `DATA_SOURCE__${
-              this.appConfig[`DATA_SOURCE_PROVIDER_${eventName}`]
-            }__${chainConfig.tag}__${eventName}`
-          ];
+          this.dataSourceUtils.getQueryEndpoint(
+            chainConfig.tag,
+            eventName as NativeTransactionKind,
+          );
       }
       this.blockchainDataSourceConfigs.push(chainConfigUpdated);
     }
+
+    console.dir(this.blockchainDataSourceConfigs, { depth: null });
   }
 
   async getOrCreateBlockchain(blockchainId: string): Promise<Blockchain> {

src/modules/queueProcessor/services/producers/accountAggregationFlow.producer.ts

@@ -28,7 +28,7 @@ export class AccountAggregationFlowProducer {
 
   async enqueueTask(args: EnqueueAccountAggregationJobInput) {
     const taskPayload: RefreshAccountTxHistoryJobDataDto = {
-      publicKey: this.cryptoUtils.addressToHex(args.publicKey),
+      publicKey: this.cryptoUtils.addressToHexIfNotHex(args.publicKey),
     };
 
     console.log(args);

src/modules/queueProcessor/services/producers/datasourceChunksParallelHandling.producer.ts

@@ -25,20 +25,30 @@ export class DatasourceChunksParallelHandlingProducer {
           'TRANSFER_CHUNK',
           requestData,
           {
-            attempts: 5,
+            attempts: 3,
             timeout: 60 * 1000,
             jobId: crypto.randomUUID(),
-            removeOnComplete: true,
+            removeOnComplete: false,
             removeOnFail: false,
             stackTraceLimit: 100,
             priority: requestData.onDemand ? 1 : 2,
           },
         );
 
-        const jobResult = await job.finished();
-        // TODO add result check
-        // TODO Add a watchdog to check if the job has finished periodically. Since pubsub does not give any guarantees.
-        resolve({ jobResult: JSON.parse(jobResult) });
+        // Watchdog to check if the job has finished periodically. Since pubsub does not give any guarantees.
+
+        const intervalInst = setInterval(async () => {
+          const jobStatus = await job.getState();
+
+          if (jobStatus === 'completed' || jobStatus === 'failed') {
+            clearInterval(intervalInst);
+            const jobRes =
+              await this.datasourceChunksParallelHandlingQueue.getJob(job.id);
+            await jobRes.remove();
+            resolve({ jobResult: JSON.parse(jobRes.returnvalue) });
+            return;
+          }
+        }, 500);
       } catch (e) {
         reject(e);
       }