Initialize compute

src/@types/C2D.ts

@@ -0,0 +1,32 @@
+import type { MetadataAlgorithm } from './DDO/Metadata.js'
+import type { Command } from './commands.js'
+export interface Compute {
+  env: string // with hash
+  validUntil: number
+}
+
+export interface ComputeAsset {
+  url?: string
+  documentId: string
+  serviceId: string
+  transferTxId?: string
+  userdata?: { [key: string]: any }
+}
+
+export interface ComputeAlgorithm {
+  documentId?: string
+  serviceId?: string
+  url?: string
+  meta?: MetadataAlgorithm
+  transferTxId?: string
+  algocustomdata?: { [key: string]: any }
+  userdata?: { [key: string]: any }
+}
+
+export interface InitializeComputeCommand extends Command {
+  datasets: [ComputeAsset]
+  algorithm: ComputeAlgorithm
+  compute: Compute
+  consumerAddress: string
+  chainId: number
+}

src/@types/DDO/Metadata.ts

@@ -23,7 +23,7 @@ export interface MetadataAlgorithm {
    * Object describing the Docker container image.
    * @type {Object}
    */
-  container: {
+  container?: {
     /**
      * The command to execute, or script to run inside the Docker image.
      * @type {string}

src/components/core/compute.ts

@@ -3,14 +3,18 @@ import { P2PCommandResponse } from '../../@types'
 import { CORE_LOGGER } from '../../utils/logging/common.js'
 import { Handler } from './handler.js'
 import { GetEnvironmentsCommand } from '../../@types/commands.js'
+import { InitializeComputeCommand } from '../../@types/C2D'
 import { getConfiguration } from '../../utils/config.js'
+import { PROTOCOL_COMMANDS } from '../../utils/constants.js'
+import { streamToString } from '../../utils/util.js'
 import axios from 'axios'
+import { validateProviderFeesForDatasets } from './utils/initializeCompute.js'
 
 export class GetEnvironmentsHandler extends Handler {
   async handle(task: GetEnvironmentsCommand): Promise<P2PCommandResponse> {
     try {
       CORE_LOGGER.logMessage(
-        'File Info Request recieved with arguments: ' + JSON.stringify(task, null, 2),
+        'Get C2D Envs Request recieved with arguments: ' + JSON.stringify(task, null, 2),
         true
       )
       const response: any[] = []
@@ -53,3 +57,85 @@ export class GetEnvironmentsHandler extends Handler {
     }
   }
 }
+
+export class InitializeComputeHandler extends Handler {
+  validateTimestamp(value: number) {
+    // in miliseconds
+    const timestampNow = new Date().getTime() / 1000
+    const validUntil = new Date(value).getTime() / 1000
+
+    return validUntil > timestampNow
+  }
+
+  checksC2DEnv(computeEnv: string, c2dEnvsWithHash: any[]): boolean {
+    for (const c of c2dEnvsWithHash) {
+      if (c.id === computeEnv) {
+        return true
+      }
+    }
+    return false
+  }
+
+  async handle(task: InitializeComputeCommand): Promise<P2PCommandResponse> {
+    try {
+      CORE_LOGGER.logMessage(
+        'Initialize Compute Request recieved with arguments: ' +
+          JSON.stringify(task, null, 2),
+        true
+      )
+
+      const { validUntil } = task.compute
+      if (!this.validateTimestamp(validUntil)) {
+        const errorMsg = `Error validating validUntil ${validUntil}. It is not in the future.`
+        CORE_LOGGER.error(errorMsg)
+        return {
+          stream: null,
+          status: {
+            httpStatus: 400,
+            error: errorMsg
+          }
+        }
+      }
+
+      const c2dEnvTask: GetEnvironmentsCommand = {
+        chainId: task.chainId,
+        command: PROTOCOL_COMMANDS.GET_COMPUTE_ENVIRONMENTS
+      }
+
+      const req = await new GetEnvironmentsHandler(this.getOceanNode()).handle(c2dEnvTask)
+
+      const resp = await streamToString(req.stream as Readable)
+      const c2dEnvs = JSON.parse(resp)
+
+      if (!this.checksC2DEnv(task.compute.env, c2dEnvs)) {
+        const errorMsg = `Compute env was not found.`
+        CORE_LOGGER.error(errorMsg)
+        return {
+          stream: null,
+          status: {
+            httpStatus: 400,
+            error: errorMsg
+          }
+        }
+      }
+      return await validateProviderFeesForDatasets(
+        this.getOceanNode(),
+        task.datasets,
+        task.algorithm,
+        task.chainId,
+        task.compute.env,
+        task.compute.validUntil,
+        task.consumerAddress
+      )
+    } catch (error) {
+      CORE_LOGGER.error(error.message)
+      return {
+        stream: null,
+        status: {
+          httpStatus: 500,
+          error: error.message
+        }
+      }
+    }
+  }
+}

src/components/core/coreHandlersRegistry.ts

@@ -18,7 +18,7 @@ import { StatusHandler } from './statusHandler.js'
 import { ReindexHandler } from './reindexHandler.js'
 import { OceanNode } from '../../OceanNode.js'
 import { Command } from '../../@types/commands.js'
-import { GetEnvironmentsHandler } from './compute.js'
+import { GetEnvironmentsHandler, InitializeComputeHandler } from './compute.js'
 
 export type HandlerRegistry = {
   handlerName: string // name of the handler
@@ -74,6 +74,10 @@ export class CoreHandlersRegistry {
       PROTOCOL_COMMANDS.GET_COMPUTE_ENVIRONMENTS,
       new GetEnvironmentsHandler(node)
     )
+    this.registerCoreHandler(
+      PROTOCOL_COMMANDS.INITIALIZE_COMPUTE,
+      new InitializeComputeHandler(node)
+    )
   }
 
   public static getInstance(node: OceanNode): CoreHandlersRegistry {

src/components/core/utils/feesHandler.ts

@@ -14,11 +14,11 @@ import { verifyMessage } from '../../../utils/blockchain.js'
 import { getConfiguration } from '../../../utils/config.js'
 import { CORE_LOGGER } from '../../../utils/logging/common.js'
 import { LOG_LEVELS_STR } from '../../../utils/logging/Logger.js'
-import { findEventByKey } from '../../Indexer/utils.js'
 import axios from 'axios'
 import { getOceanArtifactsAdresses } from '../../../utils/address.js'
 import ERC20Template from '@oceanprotocol/contracts/artifacts/contracts/templates/ERC20TemplateEnterprise.sol/ERC20TemplateEnterprise.json' assert { type: 'json' }
 import { C2DClusterInfo } from '../../../@types'
+import { verifyComputeProviderFees } from '../validateTransaction.js'
 
 export async function getC2DEnvs(asset: DDO): Promise<Array<any>> {
   try {
@@ -159,56 +159,47 @@ export async function calculateComputeProviderFee(
   return providerFee
 }
 
-async function getEventData(
-  provider: JsonRpcApiProvider,
-  transactionHash: string,
-  abi: any
-): Promise<ethers.LogDescription> {
-  const iface = new ethers.Interface(abi)
-  const receipt = await provider.getTransactionReceipt(transactionHash)
-  const eventObj = {
-    topics: receipt.logs[0].topics as string[],
-    data: receipt.logs[0].data
-  }
-  return iface.parseLog(eventObj)
-}
-// It will be used in #230 for initializeCompute
 export async function validateComputeProviderFee(
   provider: JsonRpcApiProvider,
   tx: string,
   computeEnv: string, // with hash
   asset: DDO,
   service: Service,
-  validUntil: number
+  validUntil: number,
+  userAddress: string
 ): Promise<[boolean, ProviderFeeData | {}]> {
-  const txReceipt = await provider.getTransactionReceipt(tx)
-  const { logs } = txReceipt
-  const timestampNow = new Date().getTime() / 1000
-  for (const log of logs) {
-    const event = findEventByKey(log.topics[0])
-    if (event && event.type === 'ProviderFee') {
-      const decodedEventData = await getEventData(provider, tx, ERC20Template.abi)
-      const validUntilContract = parseInt(decodedEventData.args[7].toString())
-      if (timestampNow >= validUntilContract) {
-        // provider fee expired -> reuse order
-        CORE_LOGGER.log(
-          LOG_LEVELS_STR.LEVEL_INFO,
-          `Provider fees for this env have expired -> reuse order.`,
-          true
-        )
-        const envId = computeEnv.split('-')[1]
-        const newProviderFee = await calculateComputeProviderFee(
-          asset,
-          validUntil,
-          envId,
-          service,
-          provider
-        )
-        return [false, newProviderFee]
-      } else {
-        return [true, {}]
-      }
+  try {
+    const timestampNow = new Date().getTime() / 1000
+    const validationResult = await verifyComputeProviderFees(
+      tx,
+      userAddress,
+      provider,
+      timestampNow,
+      service.timeout
+    )
+
+    if (validationResult.isValid === false) {
+      // provider fee expired or tx id is not provided -> reuse order
+      CORE_LOGGER.log(
+        LOG_LEVELS_STR.LEVEL_INFO,
+        `${validationResult.message} -> create new provider fees.`,
+        true
+      )
+      const regex = /[^-]*-(ocean-[^-]*)/
+      const envId = computeEnv.match(regex)[1]
+      const newProviderFee = await calculateComputeProviderFee(
+        asset,
+        validUntil,
+        envId,
+        service,
+        provider
+      )
+      return [false, newProviderFee]
+    } else {
+      return [true, validationResult.message]
     }
+  } catch (err) {
+    CORE_LOGGER.logMessage(`Validation for compute provider fees failed due to: ${err}`)
   }
 }
 // equiv to get_provider_fees

src/components/core/utils/initializeCompute.ts

@@ -0,0 +1,92 @@
+import { P2PCommandResponse } from '../../../@types'
+import { ComputeAlgorithm, ComputeAsset } from '../../../@types/C2D.js'
+import { CORE_LOGGER } from '../../../utils/logging/common.js'
+import { DDO } from '../../../@types/DDO/DDO.js'
+import { getJsonRpcProvider } from '../../../utils/blockchain.js'
+import { validateComputeProviderFee } from './feesHandler.js'
+import { Readable } from 'stream'
+import { OceanNode } from '../../../OceanNode'
+import { Service } from '../../../@types/DDO/Service.js'
+
+export function getServiceById(ddo: DDO, serviceId: string): Service {
+  try {
+    return ddo.services.filter((service) => service.id === serviceId)[0]
+  } catch (err) {
+    CORE_LOGGER.error(`Service was not found: ${err}`)
+  }
+}
+
+export async function validateProviderFeesForDatasets(
+  node: OceanNode,
+  datasets: [ComputeAsset],
+  algorithm: ComputeAlgorithm,
+  chainId: number,
+  env: string,
+  validUntil: number,
+  consumerAddress: string
+): Promise<P2PCommandResponse> {
+  const listOfAssest = [...datasets, ...[algorithm]]
+  const approvedParams: any = {
+    algorithm: {},
+    datasets: []
+  }
+  const provider = await getJsonRpcProvider(chainId)
+
+  for (const asset of listOfAssest) {
+    try {
+      const ddo = (await node.getDatabase().ddo.retrieve(asset.documentId)) as DDO
+      if (ddo.id === algorithm.documentId) {
+        if (ddo.metadata.type !== 'algorithm') {
+          const errorMsg = `DID is not a valid algorithm`
+          CORE_LOGGER.error(errorMsg)
+          return {
+            stream: null,
+            status: {
+              httpStatus: 400,
+              error: errorMsg
+            }
+          }
+        }
+      }
+      const service = getServiceById(ddo, asset.serviceId)
+
+      const resultValidation = await validateComputeProviderFee(
+        provider,
+        asset.transferTxId,
+        env,
+        ddo,
+        service,
+        validUntil,
+        consumerAddress
+      )
+      if (ddo.metadata.type === 'algorithm') {
+        approvedParams.algorithm = {
+          datatoken: service.datatokenAddress,
+          providerFee: resultValidation[1],
+          validOrder: resultValidation[0]
+        }
+      } else {
+        approvedParams.datasets.push({
+          datatoken: service.datatokenAddress,
+          providerFee: resultValidation[1],
+          validOrder: resultValidation[0]
+        })
+      }
+    } catch (error) {
+      CORE_LOGGER.error(`Unable to get compute provider fees: ${error}`)
+    }
+  }
+
+  const result = JSON.stringify(approvedParams, (key, value) => {
+    if (typeof value === 'bigint') {
+      return value.toString()
+    }
+    return value
+  })
+  return {
+    stream: Readable.from(result),
+    status: {
+      httpStatus: 200
+    }
+  }
+}