Feat/kyve storage provider (#60)

* feat: implemented kyve-storage-provider dedicated for dev- and testnet

common/protocol/src/index.ts

@@ -44,7 +44,6 @@ import {
   waitForCacheContinuation,
   waitForNextBundleProposal,
   waitForUploadInterval,
-  getProxyAuth,
   isStorageBalanceLow,
 } from "./methods";
 import { ICacheProvider, IMetrics, IRuntime } from "./types";
@@ -124,7 +123,6 @@ export class Validator {
   // helpers
   protected archiveDebugBundle = archiveDebugBundle;
   protected continueRound = continueRound;
-  public getProxyAuth = getProxyAuth;
 
   // txs
   protected claimUploaderRole = claimUploaderRole;
@@ -152,9 +150,9 @@ export class Validator {
   protected runCache = runCache;
 
   // factories
-  public static cacheProviderFactory = cacheProviderFactory;
-  public static storageProviderFactory = storageProviderFactory;
-  public static compressionFactory = compressionFactory;
+  protected cacheProviderFactory = cacheProviderFactory;
+  protected storageProviderFactory = storageProviderFactory;
+  protected compressionFactory = compressionFactory;
 
   /**
    * Constructor for the validator class. It is required to provide the
@@ -315,6 +313,9 @@ export class Validator {
       process.exit(1);
     }
 
+    await this.setupValidator();
+    await this.setupCacheProvider();
+
     if (await this.isStorageBalanceZero()) {
       process.exit(1);
     }
@@ -323,9 +324,6 @@ export class Validator {
       process.exit(1);
     }
 
-    await this.setupValidator();
-    await this.setupCacheProvider();
-
     // start the node process. Validator and cache should run at the same time.
     // Thats why, although they are async they are called synchronously
     try {

common/protocol/src/methods/checks/isStorageBalanceLow.ts

@@ -11,10 +11,8 @@ import { Validator, standardizeJSON } from "../..";
  */
 export async function isStorageBalanceLow(this: Validator): Promise<void> {
   try {
-    const storageProvider = Validator.storageProviderFactory(
-      this.pool.data?.current_storage_provider_id ?? 0,
-      this.storagePriv
-    );
+    this.logger.debug(`this.storageProviderFactory()`);
+    const storageProvider = this.storageProviderFactory();
 
     this.logger.info(
       `Checking account balance on StorageProvider:${storageProvider.name}`

common/protocol/src/methods/checks/isStorageBalanceZero.ts

@@ -11,10 +11,8 @@ import { Validator, standardizeError } from "../..";
  */
 export async function isStorageBalanceZero(this: Validator): Promise<boolean> {
   try {
-    const storageProvider = Validator.storageProviderFactory(
-      this.pool.data?.current_storage_provider_id ?? 0,
-      this.storagePriv
-    );
+    this.logger.debug(`this.storageProviderFactory()`);
+    const storageProvider = this.storageProviderFactory();
 
     this.logger.info(
       `Checking account balance on StorageProvider:${storageProvider.name}`

common/protocol/src/methods/index.ts

@@ -23,7 +23,6 @@ export * from "./timeouts/waitForUploadInterval";
 // helpers
 export * from "./helpers/archiveDebugBundle";
 export * from "./helpers/continueRound";
-export * from "./helpers/getProxyAuth";
 
 // txs
 export * from "./txs/claimUploaderRole";

common/protocol/src/methods/queries/getBalancesForMetrics.ts

@@ -70,15 +70,8 @@ export async function getBalancesForMetrics(this: Validator): Promise<void> {
 
   try {
     // get current storage provider defined on pool
-    this.logger.debug(
-      `storageProviderFactory(${
-        this.pool.data?.current_storage_provider_id ?? 0
-      }, $STORAGE_PRIV)`
-    );
-    const storageProvider = Validator.storageProviderFactory(
-      this.pool.data?.current_storage_provider_id ?? 0,
-      this.storagePriv
-    );
+    this.logger.debug(`this.storageProviderFactory()`);
+    const storageProvider = this.storageProviderFactory();
 
     this.logger.debug(`this.storageProvider.getBalance()`);
 

common/protocol/src/methods/setups/setupCacheProvider.ts

@@ -19,7 +19,7 @@ export async function setupCacheProvider(this: Validator): Promise<void> {
     this.logger.debug(`Initializing cache provider with path ${cachePath}`);
 
     // create cache provider depending on chosen cache type
-    this.cacheProvider = Validator.cacheProviderFactory(this.cache);
+    this.cacheProvider = this.cacheProviderFactory();
 
     // delete all contents of cache directory
     await fse.emptyDir(`${cachePath}/`);

common/protocol/src/methods/upload/createBundleProposal.ts

@@ -104,12 +104,8 @@ export async function createBundleProposal(this: Validator): Promise<void> {
     }
 
     // get current compression defined on pool
-    this.logger.debug(
-      `compressionFactory(${this.pool.data?.current_compression_id ?? 0})`
-    );
-    const compression = Validator.compressionFactory(
-      this.pool.data?.current_compression_id ?? 0
-    );
+    this.logger.debug(`this.compressionFactory()`);
+    const compression = this.compressionFactory();
 
     const uploadBundle = bundleToBytes(bundleProposal);
 
@@ -209,15 +205,8 @@ export async function createBundleProposal(this: Validator): Promise<void> {
     // uploader role to prevent upload slashes
     try {
       // get current storage provider defined on pool
-      this.logger.debug(
-        `storageProviderFactory(${
-          this.pool.data?.current_storage_provider_id ?? 0
-        }, $STORAGE_PRIV)`
-      );
-      const storageProvider = Validator.storageProviderFactory(
-        this.pool.data?.current_storage_provider_id ?? 0,
-        this.storagePriv
-      );
+      this.logger.debug(`this.storageProviderFactory()`);
+      const storageProvider = this.storageProviderFactory();
 
       // if balance is less than the upload cost we skip the uploader
       // role with a warning

common/protocol/src/methods/validate/saveBundleDecompress.ts

@@ -17,12 +17,8 @@ export async function saveBundleDecompress(
 ): Promise<DataItem[]> {
   try {
     // get compression the proposed bundle was compressed with
-    this.logger.debug(
-      `compressionFactory(${this.pool.bundle_proposal?.compression_id ?? 0})`
-    );
-    const compression = Validator.compressionFactory(
-      this.pool.bundle_proposal?.compression_id ?? 0
-    );
+    this.logger.debug(`this.compressionFactory()`);
+    const compression = this.compressionFactory();
 
     this.logger.debug(`this.compression.decompress($RAW_STORAGE_DATA)`);
 

common/protocol/src/methods/validate/saveBundleDownload.ts

@@ -53,15 +53,8 @@ export async function saveBundleDownload(
       }
 
       // get storage provider the proposed bundle was saved to
-      this.logger.debug(
-        `storageProviderFactory(${
-          this.pool.bundle_proposal?.storage_provider_id ?? 0
-        }, $STORAGE_PRIV)`
-      );
-      const storageProvider = Validator.storageProviderFactory(
-        this.pool.bundle_proposal?.storage_provider_id ?? 0,
-        this.storagePriv
-      );
+      this.logger.debug(`this.storageProviderFactory()`);
+      const storageProvider = this.storageProviderFactory();
 
       // calculate download timeout for storage provider
       // the timeout should always be 20 seconds less than the upload interval

common/protocol/src/reactors/cacheProvider/index.ts

@@ -1,4 +1,4 @@
-import { ICacheProvider } from "../..";
+import { ICacheProvider, Validator } from "../..";
 import { JsonFileCache } from "./JsonFileCache";
 import { MemoryCache } from "./MemoryCache";
 
@@ -10,14 +10,13 @@ import { MemoryCache } from "./MemoryCache";
  * x - Memory (default)
  *
  * @method cacheProviderFactory
- * @param {string} cacheId the id of the compression
  * @return {ICacheProvider}
  */
-export const cacheProviderFactory = (cacheId: string): ICacheProvider => {
-  switch (cacheId) {
+export function cacheProviderFactory(this: Validator): ICacheProvider {
+  switch (this.cache) {
     case "jsonfile":
       return new JsonFileCache();
     default:
       return new MemoryCache();
   }
-};
+}

common/protocol/src/reactors/compression/index.ts

@@ -1,4 +1,4 @@
-import { ICompression } from "../..";
+import { ICompression, Validator } from "../..";
 import { Gzip } from "./Gzip";
 import { NoCompression } from "./NoCompression";
 
@@ -11,14 +11,13 @@ import { NoCompression } from "./NoCompression";
  * x - NoCompression (default)
  *
  * @method compressionFactory
- * @param {number} compressionId the id of the compression
  * @return {ICompression}
  */
-export const compressionFactory = (compressionId: number): ICompression => {
-  switch (compressionId) {
+export function compressionFactory(this: Validator): ICompression {
+  switch (this.pool.data?.current_compression_id ?? 0) {
     case 1:
       return new Gzip();
     default:
       return new NoCompression();
   }
-};
+}

common/protocol/src/reactors/storageProviders/Kyve.ts

@@ -0,0 +1,109 @@
+import KyveSDK from "@kyvejs/sdk";
+import { v4 as uuidv4 } from "uuid";
+import axios from "axios";
+import { BundleTag, IStorageProvider } from "../../types";
+
+require("dotenv").config();
+
+export class Kyve implements IStorageProvider {
+  public name = "Kyve";
+  public coinDecimals = 0;
+
+  private chainId: string;
+  private poolId: number;
+  private staker: string;
+  private valaccount: string;
+
+  constructor(
+    chainId: string,
+    poolId: number,
+    staker: string,
+    valaccount: string
+  ) {
+    if (!chainId) {
+      throw new Error("ChainId is empty.");
+    }
+
+    if (poolId === null || poolId === undefined) {
+      throw new Error("PoolId is empty.");
+    }
+
+    if (!staker) {
+      throw new Error("Staker is empty.");
+    }
+
+    if (!valaccount) {
+      throw new Error("Valaccount mnemonic is empty.");
+    }
+
+    this.chainId = chainId;
+    this.poolId = poolId;
+    this.staker = staker;
+    this.valaccount = valaccount;
+  }
+
+  async getAddress() {
+    return (await new KyveSDK().fromMnemonic(this.valaccount)).account.address;
+  }
+
+  async getBalance() {
+    return "";
+  }
+
+  async getPrice(_: number) {
+    return "0";
+  }
+
+  async saveBundle(bundle: Buffer, _: BundleTag[]) {
+    const storageId = uuidv4();
+    const sdk = await new KyveSDK().fromMnemonic(this.valaccount);
+    const address = await this.getAddress();
+    const timestamp = new Date().valueOf().toString();
+
+    const { signature, pub_key } = await sdk.signString(
+      JSON.stringify({
+        chainId: this.chainId,
+        poolId: this.poolId.toString(),
+        staker: this.staker,
+        valaccount: address,
+        timestamp,
+      })
+    );
+
+    await axios.post(
+      "https://upload.storage.kyve.network/upload",
+      {
+        name: storageId,
+        data: bundle.toString("base64"),
+      },
+      {
+        headers: {
+          "kyve-api-key": process.env.KYVE_STORAGE_PROVIDER_API_KEY || "",
+          "kyve-chain-id": this.chainId,
+          "kyve-pool-id": this.poolId.toString(),
+          "kyve-staker": this.staker,
+          "kyve-public-key": pub_key.value,
+          "kyve-timestamp": timestamp,
+          "kyve-signature": signature,
+        },
+      }
+    );
+
+    return {
+      storageId: storageId,
+      storageData: bundle,
+    };
+  }
+
+  async retrieveBundle(storageId: string, timeout: number) {
+    const { data: storageData } = await axios.get(
+      `https://storage.kyve.network/${storageId}`,
+      { responseType: "arraybuffer", timeout }
+    );
+
+    return {
+      storageId,
+      storageData,
+    };
+  }
+}