compute start (#1196)

* compute start

* add getStatus

src/@types/Compute.ts

@@ -37,17 +37,18 @@ export interface ComputeOutput {
   whitelist?: string[]
 }
 
-export interface ComputeInput {
+export interface ComputeAsset {
   documentId: string
-  serviceId: number
+  serviceId: string
   transferTxId?: string
+  userdata?: { [key: string]: any }
 }
 
 export interface ComputeAlgorithm {
-  did?: string
+  documentId?: string
-  serviceIndex?: number
+  serviceId?: string
   meta?: MetadataAlgorithm
   transferTxId?: string
-  dataToken?: string
+  algocustomdata?: { [key: string]: any }
-  algoCustomParameters?: { [key: string]: any }
+  userdata?: { [key: string]: any }
 }

src/@types/DDO/Metadata.ts

@@ -11,6 +11,12 @@ export interface MetadataAlgorithm {
    */
   version?: string
 
+  /**
+   * Rawcode
+   * @type {string}
+   */
+  rawcode?: string
+
   /**
    * Object describing the Docker container image.
    * @type {Object}

src/provider/Provider.ts

@@ -6,6 +6,7 @@ import {
   ComputeJob,
   ComputeOutput,
   ComputeAlgorithm,
+  ComputeAsset,
   ProviderInitialize
 } from '../@types/'
 import { noZeroX } from '../utils/ConversionTypeHelper'
@@ -229,7 +230,8 @@ export class Provider {
     consumerAddress: string,
     providerUri: string,
     getMethod: any,
-    userCustomParameters?: UserCustomParameters
+    userCustomParameters?: UserCustomParameters,
+    computeEnv?: string
   ): Promise<ProviderInitialize> {
     const providerEndpoints = await this.getEndpoints(providerUri)
     const serviceEndpoints = await this.getServiceEndpoints(
@@ -247,6 +249,7 @@ export class Provider {
     initializeUrl += `&consumerAddress=${consumerAddress}`
     if (userCustomParameters)
       initializeUrl += '&userdata=' + encodeURI(JSON.stringify(userCustomParameters))
+    if (computeEnv) initializeUrl += '&computeEnv=' + encodeURI(computeEnv)
     try {
       const response = await getMethod('GET', initializeUrl)
       const results: ProviderInitialize = await response.json()
@@ -314,12 +317,14 @@ export class Provider {
    * @return {Promise<ComputeJob | ComputeJob[]>}
    */
   public async computeStart(
-    did: string,
-    consumerAddress: string,
-    algorithm: ComputeAlgorithm,
     providerUri: string,
     web3: Web3,
     fetchMethod: any,
+    consumerAddress: string,
+    computeEnv: string,
+    dataset: ComputeAsset,
+    algorithm: ComputeAlgorithm,
+    additionalDatasets?: ComputeAsset[],
     output?: ComputeOutput
   ): Promise<ComputeJob | ComputeJob[]> {
     const providerEndpoints = await this.getEndpoints(providerUri)
@@ -331,16 +336,9 @@ export class Provider {
       ? this.getEndpointURL(serviceEndpoints, 'computeStart').urlPath
       : null
 
-    const nonce = await this.getNonce(
+    const nonce = Date.now()
-      providerUri,
-      consumerAddress,
-      fetchMethod,
-      providerEndpoints,
-      serviceEndpoints
-    )
-
     let signatureMessage = consumerAddress
-    signatureMessage += (did && `${noZeroX(did)}`) || ''
+    signatureMessage += dataset.documentId
     signatureMessage += nonce
     const signature = await this.createHashSignature(
       web3,
@@ -351,14 +349,22 @@ export class Provider {
     const payload = Object()
     payload.consumerAddress = consumerAddress
     payload.signature = signature
-    payload.algorithmDid = algorithm.did
+    payload.nonce = nonce
-    payload.algorithmMeta = algorithm.meta
+    payload.computeEnv = computeEnv
-    payload.algorithmServiceId = algorithm.serviceIndex
+    payload.dataset = dataset
+    payload.algorithm = algorithm
+    if (payload.additionalDatasets) payload.additionalDatasets = additionalDatasets
     if (output) payload.output = output
-
     if (!computeStartUrl) return null
     try {
-      const response = await fetchMethod(computeStartUrl, JSON.stringify(payload))
+      const response = await fetchMethod(
+        'POST',
+        computeStartUrl,
+        JSON.stringify(payload),
+        {
+          'Content-Type': 'application/json'
+        }
+      )
       if (response?.ok) {
         const params = await response.json()
         return params
@@ -452,13 +458,15 @@ export class Provider {
    * @return {Promise<ComputeJob | ComputeJob[]>}
    */
   public async computeStatus(
-    did: string,
-    consumerAddress: string,
     providerUri: string,
-    web3: Web3,
     fetchMethod: any,
-    jobId?: string
+    jobId?: string,
+    did?: string,
+    consumerAddress?: string
   ): Promise<ComputeJob | ComputeJob[]> {
+    if (!jobId && !did && !consumerAddress) {
+      throw new Error('You need at least one of jobId, did, consumerAddress')
+    }
     const providerEndpoints = await this.getEndpoints(providerUri)
     const serviceEndpoints = await this.getServiceEndpoints(
       providerUri,
@@ -468,32 +476,13 @@ export class Provider {
       ? this.getEndpointURL(serviceEndpoints, 'computeStatus').urlPath
       : null
 
-    const nonce = await this.getNonce(
-      providerUri,
-      consumerAddress,
-      fetchMethod,
-      providerEndpoints,
-      serviceEndpoints
-    )
-
-    let signatureMessage = consumerAddress
-    signatureMessage += jobId || ''
-    signatureMessage += (did && `${noZeroX(did)}`) || ''
-    signatureMessage += nonce
-    const signature = await this.createHashSignature(
-      web3,
-      consumerAddress,
-      signatureMessage
-    )
-
     let url = '?documentId=' + noZeroX(did)
     url += `&consumerAddress=${consumerAddress}`
-    url += (signature && `&signature=${signature}`) || ''
     url += (jobId && `&jobId=${jobId}`) || ''
 
     if (!computeStatusUrl) return null
     try {
-      const response = await fetchMethod(computeStatusUrl + url)
+      const response = await fetchMethod('GET', computeStatusUrl + url)
       if (response?.ok) {
         const params = await response.json()
         return params

test/integration/ComputeFlow.test.ts

@@ -0,0 +1,328 @@
+import config from './config'
+import ProviderInstance, { Provider } from '../../src/provider/Provider'
+import Aquarius from '../../src/aquarius/Aquarius'
+import { assert } from 'chai'
+import { NftFactory, NftCreateData } from '../../src/factories/index'
+import { Datatoken } from '../../src/tokens/Datatoken'
+import { Erc20CreateParams } from '../../src/interfaces'
+import { getHash } from '../../src/utils'
+import { Nft } from '../../src/tokens/NFT'
+import Web3 from 'web3'
+import { algo, SHA256 } from 'crypto-js'
+import { homedir } from 'os'
+import fs from 'fs'
+import { downloadFile, crossFetchGeneric } from '../../src/utils/FetchHelper'
+import console from 'console'
+
+const data = JSON.parse(
+  fs.readFileSync(
+    process.env.ADDRESS_FILE ||
+      `${homedir}/.ocean/ocean-contracts/artifacts/address.json`,
+    'utf8'
+  )
+)
+
+const addresses = data.development
+console.log(addresses)
+const aquarius = new Aquarius('http://127.0.0.1:5000')
+const web3 = new Web3('http://127.0.0.1:8545')
+const providerUrl = 'http://172.15.0.4:8030'
+const assetUrl = [
+  {
+    type: 'url',
+    url: 'https://raw.githubusercontent.com/oceanprotocol/testdatasets/main/shs_dataset_test.txt',
+    method: 'GET'
+  }
+]
+const ddo = {
+  '@context': ['https://w3id.org/did/v1'],
+  id: 'did:op:efba17455c127a885ec7830d687a8f6e64f5ba559f8506f8723c1f10f05c049c',
+  version: '4.0.0',
+  chainId: 4,
+  nftAddress: '0x0',
+  metadata: {
+    created: '2021-12-20T14:35:20Z',
+    updated: '2021-12-20T14:35:20Z',
+    type: 'dataset',
+    name: 'dfgdfgdg',
+    description: 'd dfgd fgd dfg dfgdfgd dfgdf',
+    tags: [''],
+    author: 'dd',
+    license: 'https://market.oceanprotocol.com/terms',
+    additionalInformation: {
+      termsAndConditions: true
+    }
+  },
+  services: [
+    {
+      id: 'notAnId',
+      type: 'compute',
+      files: '',
+      datatokenAddress: '0xa15024b732A8f2146423D14209eFd074e61964F3',
+      serviceEndpoint: 'https://providerv4.rinkeby.oceanprotocol.com',
+      timeout: 0,
+      compute: {
+        publisherTrustedAlgorithmPublishers: [],
+        publisherTrustedAlgorithms: [],
+        allowRawAlgorithm: true,
+        allowNetworkAccess: true,
+        namespace: 'ocean-compute',
+        cpus: 2,
+        gpus: 4,
+        gpuType: 'NVIDIA Tesla V100 GPU',
+        memory: '128M',
+        volumeSize: '2G'
+      }
+    }
+  ]
+}
+const algoAssetUrl = [
+  {
+    type: 'url',
+    url: 'https://raw.githubusercontent.com/oceanprotocol/test-algorithm/master/javascript/algo.js',
+    method: 'GET'
+  }
+]
+const algoDdo = {
+  '@context': ['https://w3id.org/did/v1'],
+  id: 'did:op:efba17455c127a885ec7830d687a8f6e64f5ba559f8506f8723c1f10f05c049c',
+  version: '4.0.0',
+  chainId: 4,
+  nftAddress: '0x0',
+  metadata: {
+    created: '2021-12-20T14:35:20Z',
+    updated: '2021-12-20T14:35:20Z',
+    type: 'algorithm',
+    name: 'dfgdfgdg',
+    description: 'd dfgd fgd dfg dfgdfgd dfgdf',
+    tags: [''],
+    author: 'dd',
+    license: 'https://market.oceanprotocol.com/terms',
+    additionalInformation: {
+      termsAndConditions: true
+    },
+    algorithm: {
+      language: 'Node.js',
+      version: '1.0.0',
+      container: {
+        entrypoint: 'node $ALGO',
+        image: 'ubuntu',
+        tag: 'latest',
+        checksum: '44e10daa6637893f4276bb8d7301eb35306ece50f61ca34dcab550'
+      }
+    }
+  },
+  services: [
+    {
+      id: 'notAnId',
+      type: 'access',
+      files: '',
+      datatokenAddress: '0xa15024b732A8f2146423D14209eFd074e61964F3',
+      serviceEndpoint: 'https://providerv4.rinkeby.oceanprotocol.com',
+      timeout: 0
+    }
+  ]
+}
+
+describe('Simple compute tests', async () => {
+  it('should publish a dataset, algorithm and start a compute job', async () => {
+    const nft = new Nft(web3)
+    const datatoken = new Datatoken(web3)
+    const Factory = new NftFactory(addresses.ERC721Factory, web3)
+    const accounts = await web3.eth.getAccounts()
+    const publisherAccount = accounts[0]
+    const consumerAccount = accounts[1]
+    const chain = await web3.eth.getChainId()
+    const nftParamsAsset: NftCreateData = {
+      name: 'testNFT',
+      symbol: 'TST',
+      templateIndex: 1,
+      tokenURI: ''
+    }
+    const erc20ParamsAsset: Erc20CreateParams = {
+      templateIndex: 1,
+      cap: '100000',
+      feeAmount: '0',
+      feeManager: '0x0000000000000000000000000000000000000000',
+      feeToken: '0x0000000000000000000000000000000000000000',
+      minter: publisherAccount,
+      mpFeeAddress: '0x0000000000000000000000000000000000000000'
+    }
+    const result = await Factory.createNftWithErc(
+      publisherAccount,
+      nftParamsAsset,
+      erc20ParamsAsset
+    )
+    const erc721AddressAsset = result.events.NFTCreated.returnValues[0]
+    const datatokenAddressAsset = result.events.TokenCreated.returnValues[0]
+
+    // create the files encrypted string
+    let providerResponse = await ProviderInstance.encrypt(
+      assetUrl,
+      providerUrl,
+      crossFetchGeneric
+    )
+    ddo.services[0].files = await providerResponse.text()
+    ddo.services[0].datatokenAddress = datatokenAddressAsset
+    // update ddo and set the right did
+    ddo.nftAddress = erc721AddressAsset
+    ddo.id =
+      'did:op:' +
+      SHA256(web3.utils.toChecksumAddress(erc721AddressAsset) + chain.toString(10))
+
+    providerResponse = await ProviderInstance.encrypt(ddo, providerUrl, crossFetchGeneric)
+    let encryptedResponse = await providerResponse.text()
+    let metadataHash = getHash(JSON.stringify(ddo))
+    let res = await nft.setMetadata(
+      erc721AddressAsset,
+      publisherAccount,
+      0,
+      providerUrl,
+      '',
+      '0x2',
+      encryptedResponse,
+      '0x' + metadataHash
+    )
+
+    // let's publish the algorithm as well
+    const nftParamsAlgo: NftCreateData = {
+      name: 'testNFT',
+      symbol: 'TST',
+      templateIndex: 1,
+      tokenURI: ''
+    }
+    const erc20ParamsAlgo: Erc20CreateParams = {
+      templateIndex: 1,
+      cap: '100000',
+      feeAmount: '0',
+      feeManager: '0x0000000000000000000000000000000000000000',
+      feeToken: '0x0000000000000000000000000000000000000000',
+      minter: publisherAccount,
+      mpFeeAddress: '0x0000000000000000000000000000000000000000'
+    }
+    const resultAlgo = await Factory.createNftWithErc(
+      publisherAccount,
+      nftParamsAlgo,
+      erc20ParamsAlgo
+    )
+    const erc721AddressAlgo = resultAlgo.events.NFTCreated.returnValues[0]
+    const datatokenAddressAlgo = resultAlgo.events.TokenCreated.returnValues[0]
+
+    // create the files encrypted string
+    providerResponse = await ProviderInstance.encrypt(
+      algoAssetUrl,
+      providerUrl,
+      crossFetchGeneric
+    )
+    algoDdo.services[0].files = await providerResponse.text()
+    algoDdo.services[0].datatokenAddress = datatokenAddressAlgo
+    // update ddo and set the right did
+    algoDdo.nftAddress = erc721AddressAlgo
+
+    algoDdo.id =
+      'did:op:' +
+      SHA256(web3.utils.toChecksumAddress(erc721AddressAlgo) + chain.toString(10))
+
+    providerResponse = await ProviderInstance.encrypt(
+      algoDdo,
+      providerUrl,
+      crossFetchGeneric
+    )
+    encryptedResponse = await providerResponse.text()
+    metadataHash = getHash(JSON.stringify(algoDdo))
+    res = await nft.setMetadata(
+      erc721AddressAlgo,
+      publisherAccount,
+      0,
+      providerUrl,
+      '',
+      '0x2',
+      encryptedResponse,
+      '0x' + metadataHash
+    )
+    // let's wait
+    const resolvedDDOAsset = await aquarius.waitForAqua(crossFetchGeneric, ddo.id)
+    assert(resolvedDDOAsset, 'Cannot fetch DDO from Aquarius')
+    const resolvedDDOAlgo = await aquarius.waitForAqua(crossFetchGeneric, algoDdo.id)
+    assert(resolvedDDOAlgo, 'Cannot fetch DDO from Aquarius')
+    // mint 1 ERC20 and send it to the consumer
+    await datatoken.mint(datatokenAddressAsset, publisherAccount, '1', consumerAccount)
+    await datatoken.mint(datatokenAddressAlgo, publisherAccount, '1', consumerAccount)
+
+    // initialize provider orders for algo
+    const initializeDataAlgo = await ProviderInstance.initialize(
+      resolvedDDOAlgo.id,
+      resolvedDDOAlgo.services[0].id,
+      0,
+      consumerAccount,
+      providerUrl,
+      crossFetchGeneric
+    )
+    // make the payment
+    const txidAlgo = await datatoken.startOrder(
+      datatokenAddressAlgo,
+      consumerAccount,
+      initializeDataAlgo.computeAddress,
+      0,
+      initializeDataAlgo.providerFee.providerFeeAddress,
+      initializeDataAlgo.providerFee.providerFeeToken,
+      initializeDataAlgo.providerFee.providerFeeAmount,
+      initializeDataAlgo.providerFee.v,
+      initializeDataAlgo.providerFee.r,
+      initializeDataAlgo.providerFee.s,
+      initializeDataAlgo.providerFee.providerData
+    )
+    assert(txidAlgo, 'Failed to order algo')
+    // initialize provider orders for asset
+    const initializeData = await ProviderInstance.initialize(
+      resolvedDDOAsset.id,
+      resolvedDDOAsset.services[0].id,
+      0,
+      consumerAccount,
+      providerUrl,
+      crossFetchGeneric,
+      null,
+      'env1'
+    )
+    // make the payment
+    const txidAsset = await datatoken.startOrder(
+      datatokenAddressAsset,
+      consumerAccount,
+      initializeDataAlgo.computeAddress,
+      0,
+      initializeData.providerFee.providerFeeAddress,
+      initializeData.providerFee.providerFeeToken,
+      initializeData.providerFee.providerFeeAmount,
+      initializeData.providerFee.v,
+      initializeData.providerFee.r,
+      initializeData.providerFee.s,
+      initializeData.providerFee.providerData
+    )
+    assert(txidAsset, 'Failed to order algo')
+    // start the compute job
+    const computeJobs = await ProviderInstance.computeStart(
+      providerUrl,
+      web3,
+      crossFetchGeneric,
+      consumerAccount,
+      'env1',
+      {
+        documentId: resolvedDDOAsset.id,
+        serviceId: resolvedDDOAsset.services[0].id,
+        transferTxId: txidAsset.transactionHash
+      },
+      {
+        documentId: resolvedDDOAlgo.id,
+        serviceId: resolvedDDOAlgo.services[0].id,
+        transferTxId: txidAlgo.transactionHash
+      }
+    )
+    assert(computeJobs, 'Cannot start compute job')
+    const jobStatus = await ProviderInstance.computeStatus(
+      providerUrl,
+      crossFetchGeneric,
+      computeJobs[0].jobId
+    )
+    assert(jobStatus)
+  })
+})

test/integration/SimplePublishConsumeFlow.test.ts

@@ -8,7 +8,6 @@ import { Erc20CreateParams } from '../../src/interfaces'
 import { getHash } from '../../src/utils'
 import { Nft } from '../../src/tokens/NFT'
 import Web3 from 'web3'
-import fetch from 'cross-fetch'
 import { SHA256 } from 'crypto-js'
 import { homedir } from 'os'
 import fs from 'fs'