Add kyve support

package.json

@@ -8,7 +8,6 @@
     "@actions/core": "^1.10.0",
     "@babel/preset-env": "^7.16.11",
     "@octokit/request": "^5.6.3",
-    "@types/cron-converter": "^1",
     "@types/node": "^18.16.10",
     "@types/node-fetch": "2.6.2",
     "@typescript-eslint/eslint-plugin": "^5.10.2",

packages/node/CHANGELOG.md

@@ -5,9 +5,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+### Added
+- Support for KYVE integration with supporting flags (#235)
+  - `--kyve-endpoint` (default value is `https://api-us-1.kyve.network`, To disable use `false`)
+  - `--kyve-chain-id` (default value is `kyve-1`)
+  - `--kyve-storage-url` (default value is `https://arweave.net`) 
 ### Removed
 - Unused @apollo/client dependency causing version conflicts (#253)
-
 ### Changed
 - Update dependencies and apply changes to match (#254)
 

packages/node/package.json

@@ -22,24 +22,21 @@
     "@cosmjs/cosmwasm-stargate": "^0.32.3",
     "@cosmjs/proto-signing": "^0.32.3",
     "@cosmjs/stargate": "^0.32.3",
+    "@kyvejs/protocol": "^1.0.12",
+    "@kyvejs/sdk": "^1.1.1",
     "@nestjs/common": "^9.4.0",
     "@nestjs/core": "^9.4.0",
     "@nestjs/event-emitter": "^2.0.0",
     "@nestjs/platform-express": "^9.4.0",
     "@nestjs/schedule": "^3.0.1",
     "@subql/common": "^3.5.1",
     "@subql/common-cosmos": "workspace:*",
-    "@subql/node-core": "^10.0.0",
+    "@subql/node-core": "^10.1.0",
     "@subql/types-cosmos": "workspace:*",
-    "cron-converter": "^1.0.2",
-    "eventemitter2": "^6.4.5",
     "lodash": "^4.17.21",
-    "pg": "^8.7.1",
     "protobufjs": "^6.11.4",
     "reflect-metadata": "^0.1.13",
     "rimraf": "^3.0.2",
-    "rxjs": "^7.5.2",
-    "vm2": "^3.9.9",
     "yargs": "^16.2.0"
   },
   "devDependencies": {

packages/node/src/configure/NodeConfig.ts

@@ -0,0 +1,41 @@
+// Copyright 2020-2024 SubQuery Pte Ltd authors & contributors
+// SPDX-License-Identifier: GPL-3.0
+
+import { SupportedChains } from '@kyvejs/sdk/src/constants';
+import { IConfig, NodeConfig } from '@subql/node-core';
+
+export interface ICosmosConfig extends IConfig {
+  kyveChainId: SupportedChains;
+  kyveEndpoint: string;
+  kyveStorageUrl: string;
+}
+
+export class CosmosNodeConfig extends NodeConfig<ICosmosConfig> {
+  /**
+   * This is a wrapper around the core NodeConfig to get additional properties that are provided through args or node runner options
+   * NOTE: This isn't injected anywhere, so you need to wrap the injected node config
+   *
+   * @example
+   * constructor(
+   *   nodeConfig: NodeConfig,
+   * ) {
+   *   this.nodeConfig = new CosmosNodeConfig(nodeConfig);
+   * }
+   * */
+  constructor(config: NodeConfig) {
+    // Rebuild with internal config
+    super((config as any)._config, (config as any)._isTest);
+  }
+
+  get kyveEndpoint(): string {
+    return this._config.kyveEndpoint;
+  }
+
+  get kyveChainId(): SupportedChains {
+    return this._config.kyveChainId;
+  }
+
+  get kyveStorageUrl(): string {
+    return this._config.kyveStorageUrl;
+  }
+}

packages/node/src/configure/SubqueryProject.ts

@@ -2,41 +2,46 @@
 // SPDX-License-Identifier: GPL-3.0
 
 import assert from 'assert';
+import fs from 'fs';
+import os from 'os';
+import { sep } from 'path';
+import { isMainThread } from 'worker_threads';
 import { Injectable } from '@nestjs/common';
 import { validateSemver } from '@subql/common';
 import {
   CosmosProjectNetworkConfig,
   parseCosmosProjectManifest,
-  CosmosDataSource,
   ProjectManifestV1_0_0Impl,
   isRuntimeCosmosDs,
   isCustomCosmosDs,
 } from '@subql/common-cosmos';
 import {
+  CronFilter,
   insertBlockFiltersCronSchedules,
   ISubqueryProject,
   loadProjectTemplates,
-  SubqlProjectDs,
   updateDataSourcesV1_0_0,
+  WorkerHost,
 } from '@subql/node-core';
 import { ParentProject, Reader, RunnerSpecs } from '@subql/types-core';
 import {
+  CosmosDatasource,
   CustomDatasourceTemplate,
   RuntimeDatasourceTemplate,
   CosmosHandlerKind,
+  CosmosBlockFilter,
 } from '@subql/types-cosmos';
 import { buildSchemaFromString } from '@subql/utils';
-import Cron from 'cron-converter';
 import { GraphQLSchema } from 'graphql';
 import { processNetworkConfig } from '../utils/project';
 
 const { version: packageVersion } = require('../../package.json');
 
-export type CosmosProjectDs = SubqlProjectDs<CosmosDataSource>;
-
 export type CosmosProjectDsTemplate =
-  | SubqlProjectDs<RuntimeDatasourceTemplate>
-  | SubqlProjectDs<CustomDatasourceTemplate>;
+  | RuntimeDatasourceTemplate
+  | CustomDatasourceTemplate;
+
+export type SubqlProjectBlockFilter = CosmosBlockFilter & CronFilter;
 
 const NOT_SUPPORT = (name: string) => {
   throw new Error(`Manifest specVersion ${name} is not supported`);
@@ -47,22 +52,23 @@ type NetworkConfig = CosmosProjectNetworkConfig & { chainId: string };
 
 @Injectable()
 export class SubqueryProject implements ISubqueryProject {
-  #dataSources: CosmosProjectDs[];
+  #dataSources: CosmosDatasource[];
 
   constructor(
     readonly id: string,
     readonly root: string,
     readonly network: NetworkConfig,
-    dataSources: CosmosProjectDs[],
+    dataSources: CosmosDatasource[],
     readonly schema: GraphQLSchema,
     readonly templates: CosmosProjectDsTemplate[],
     readonly runner?: RunnerSpecs,
     readonly parent?: ParentProject,
+    readonly tempDir?: string,
   ) {
     this.#dataSources = dataSources;
   }
 
-  get dataSources(): CosmosProjectDs[] {
+  get dataSources(): CosmosDatasource[] {
     return this.#dataSources;
   }
 
@@ -112,6 +118,23 @@ export class SubqueryProject implements ISubqueryProject {
 
 type SUPPORT_MANIFEST = ProjectManifestV1_0_0Impl;
 
+/**
+ * Gets a temp dir shared between main thread and workers
+ * */
+function getTempDir(): string {
+  if (isMainThread) return fs.mkdtempSync(`${os.tmpdir()}${sep}`);
+  const workerTempDir = (
+    (global as any).host as WorkerHost<any> | undefined
+  )?.getWorkerData()?.tempDir;
+
+  if (!workerTempDir) {
+    throw new Error(
+      'Worker expected tempDir to be provided through workerData',
+    );
+  }
+  return workerTempDir;
+}
+
 async function loadProjectFromManifestBase(
   projectManifest: SUPPORT_MANIFEST,
   reader: Reader,
@@ -177,5 +200,6 @@ async function loadProjectFromManifestBase(
     templates,
     runner,
     projectManifest.parent,
+    getTempDir(),
   );
 }

packages/node/src/indexer/api.service.test.ts

@@ -8,45 +8,59 @@ import { toRfc3339WithNanoseconds } from '@cosmjs/tendermint-rpc';
 import { INestApplication } from '@nestjs/common';
 import { EventEmitter2, EventEmitterModule } from '@nestjs/event-emitter';
 import { Test } from '@nestjs/testing';
-import { loadFromJsonOrYaml } from '@subql/common';
-import { ConnectionPoolService, delay, NodeConfig } from '@subql/node-core';
+import { loadFromJsonOrYaml, makeTempDir } from '@subql/common';
+import {
+  ConnectionPoolService,
+  ConnectionPoolStateManager,
+  NodeConfig,
+} from '@subql/node-core';
 import { GraphQLSchema } from 'graphql';
 import { SubqueryProject } from '../configure/SubqueryProject';
 import { ApiService } from './api.service';
 
-const ENDPOINT = 'https://rpc-juno.itastakers.com/';
-const CHAINID = 'juno-1';
-
 const TEST_BLOCKNUMBER = 3266772;
 
 const projectsDir = path.join(__dirname, '../../test');
 
-function testCosmosProject(): SubqueryProject {
+function testCosmosProject(
+  endpoint: string,
+  chainId: string,
+  fileCacheDir?: string,
+): SubqueryProject {
   return {
     network: {
-      endpoint: ENDPOINT,
-      chainId: CHAINID,
+      endpoint: [endpoint],
+      chainId: chainId,
     },
     dataSources: [],
     id: 'test',
     root: './',
     schema: new GraphQLSchema({}),
     templates: [],
+    tempDir: fileCacheDir,
   } as SubqueryProject;
 }
 
 jest.setTimeout(200000);
 
-describe.skip('ApiService', () => {
+describe('ApiService', () => {
   let app: INestApplication;
   let apiService: ApiService;
-  const prepareApiService = async () => {
+
+  let tmpPath: string;
+
+  const prepareApiService = async (
+    endpoint: string,
+    chainId: string,
+    fileCacheDir?: string,
+  ) => {
     const module = await Test.createTestingModule({
       providers: [
+        ConnectionPoolStateManager,
         ConnectionPoolService,
         {
           provide: 'ISubqueryProject',
-          useFactory: () => testCosmosProject(),
+          useFactory: () => testCosmosProject(endpoint, chainId, fileCacheDir),
         },
         {
           provide: NodeConfig,
@@ -61,38 +75,68 @@ describe.skip('ApiService', () => {
     app = module.createNestApplication();
     await app.init();
     apiService = app.get(ApiService);
+    (apiService as any).nodeConfig._config.kyveEndpoint =
+      'https://api-us-1.kyve.network';
+    (apiService as any).nodeConfig._config.kyveStorageUrl =
+      'https://arweave.net';
     await apiService.init();
   };
 
-  beforeAll(async () => {
-    await prepareApiService();
-  });
+  const ENDPOINT = 'https://rpc-juno.itastakers.com/';
+  const CHAINID = 'juno-1';
+
+  describe('RPC api service', () => {
+    beforeAll(async () => {
+      tmpPath = await makeTempDir();
+    });
+    it('Falls back on rpc if kyve fails', async () => {
+      const endpoint = 'https://rpc.mainnet.archway.io:443';
+      const chainId = 'archway-1';
+
+      await prepareApiService(endpoint, chainId, tmpPath);
 
-  it('query block info', async () => {
-    const api = apiService.api;
-    const blockInfo = await api.blockInfo(TEST_BLOCKNUMBER);
-    const doc: any = loadFromJsonOrYaml(
-      path.join(projectsDir, 'block_3266772.json'),
-    );
-    const realBlockInfo = {
-      id: toHex(doc.block_id.hash).toUpperCase(),
-      header: {
-        version: {
-          block: new Uint53(+doc.block.header.version.block).toString(),
-          app: blockInfo.block.header.version.app,
+      jest
+        .spyOn((apiService as any).kyveApi, 'retrieveBundleData')
+        .mockRejectedValueOnce(
+          'Error: Client network socket disconnected before secure TLS connection was established',
+        );
+
+      const rpcFetchSpy = jest.spyOn(apiService as any, 'retryFetch');
+
+      await apiService.fetchBlocks([4282099]);
+
+      expect(rpcFetchSpy).toHaveBeenCalledTimes(1);
+    });
+    it.skip('query block info', async () => {
+      await prepareApiService(ENDPOINT, CHAINID, tmpPath);
+
+      const api = apiService.api;
+      const blockInfo = await api.blockInfo(TEST_BLOCKNUMBER);
+      const doc: any = loadFromJsonOrYaml(
+        path.join(projectsDir, 'block_3266772.json'),
+      );
+      const realBlockInfo = {
+        id: toHex(doc.block_id.hash).toUpperCase(),
+        header: {
+          version: {
+            block: new Uint53(+doc.block.header.version.block).toString(),
+            app: blockInfo.block.header.version.app,
+          },
+          height: doc.block.header.height,
+          chainId: doc.block.header.chainId,
+          time: toRfc3339WithNanoseconds(doc.block.header.time),
         },
-        height: doc.block.header.height,
-        chainId: doc.block.header.chainId,
-        time: toRfc3339WithNanoseconds(doc.block.header.time),
-      },
-      txs: doc.block.txs,
-    };
-    expect(blockInfo).toMatchObject(realBlockInfo);
-  });
+        txs: doc.block.txs,
+      };
+      expect(blockInfo).toMatchObject(realBlockInfo);
+    });
+
+    it.skip('query tx info by height', async () => {
+      await prepareApiService(ENDPOINT, CHAINID, tmpPath);
 
-  it('query tx info by height', async () => {
-    const api = apiService.api;
-    const txInfos = await api.txInfoByHeight(TEST_BLOCKNUMBER);
-    expect(txInfos.length).toEqual(4);
+      const api = apiService.api;
+      const txInfos = await api.txInfoByHeight(TEST_BLOCKNUMBER);
+      expect(txInfos.length).toEqual(4);
+    });
   });
 });