Merge pull request #1143 from oceanprotocol/feature/provider-c2d-support

add c2d methods in provider class

src/@types/Compute.ts

@@ -0,0 +1,53 @@
+import { Metadata, MetadataAlgorithm } from './Metadata'
+
+export type ComputeResultType = 'algorithmLog' | 'output'
+
+export interface ComputeResult {
+  filename: string
+  filesize: number
+  type: ComputeResultType
+  index?: number
+}
+
+export interface ComputeJob {
+  owner: string
+  did?: string
+  jobId: string
+  dateCreated: string
+  dateFinished: string
+  status: number
+  statusText: string
+  results: ComputeResult[]
+  inputDID?: string[]
+  algoDID?: string
+  agreementId?: string
+  expireTimestamp: number
+}
+
+export interface ComputeOutput {
+  publishAlgorithmLog?: boolean
+  publishOutput?: boolean
+  providerAddress?: string
+  providerUri?: string
+  metadata?: Metadata
+  metadataUri?: string
+  nodeUri?: string
+  owner?: string
+  secretStoreUri?: string
+  whitelist?: string[]
+}
+
+export interface ComputeInput {
+  documentId: string
+  serviceId: number
+  transferTxId?: string
+}
+
+export interface ComputeAlgorithm {
+  did?: string
+  serviceIndex?: number
+  meta?: MetadataAlgorithm
+  transferTxId?: string
+  dataToken?: string
+  algoCustomParameters?: { [key: string]: any }
+}

src/@types/index.ts

@@ -4,3 +4,4 @@ export * from './Service'
 export * from './Credentials'
 export * from './Metadata'
 export * from './FileMetadata'
+export * from './Compute'

src/provider/Provider.ts

@@ -1,8 +1,14 @@
 import Web3 from 'web3'
 import { LoggerInstance } from '../utils'
-import { Asset, FileMetadata } from '../@types/'
+import {
+  Asset,
+  FileMetadata,
+  ComputeJob,
+  ComputeOutput,
+  ComputeAlgorithm
+} from '../@types/'
 import { noZeroX } from '../utils/ConversionTypeHelper'
-import { signText } from '../utils/SignatureUtils'
+import { signText, signWithHash } from '../utils/SignatureUtils'
 
 export interface ServiceEndpoint {
   serviceName: string
@@ -89,6 +95,24 @@ export class Provider {
     }
   }
 
+  public async createSignature(
+    web3: Web3,
+    accountId: string,
+    agreementId: string
+  ): Promise<string> {
+    const signature = await signText(web3, noZeroX(agreementId), accountId)
+    return signature
+  }
+
+  public async createHashSignature(
+    web3: Web3,
+    accountId: string,
+    message: string
+  ): Promise<string> {
+    const signature = await signWithHash(web3, message, accountId)
+    return signature
+  }
+
   /** Encrypt DDO using the Provider's own symmetric key
    * @param {string} did Identifier of the asset to be registered in ocean
    * @param {string} accountId Publisher address
@@ -206,6 +230,18 @@ export class Provider {
     }
   }
 
+  /** Allows download of asset data file.
+   * @param {string} did
+   * @param {string} destination
+   * @param {string} accountId
+   * @param {FileMetadata[]} files
+   * @param {-1} index
+   * @param {string} providerUri
+   * @param {Web3} web3
+   * @param {any} fetchMethod
+   * @param {UserCustomParameters} userCustomParameters
+   * @return {Promise<any>}
+   */
   public async download(
     did: string,
     destination: string,
@@ -260,13 +296,344 @@ export class Provider {
     return destination
   }
 
-  public async createSignature(
+  /** Instruct the provider to start a compute job
+   * @param {string} did
+   * @param {string} consumerAddress
+   * @param {ComputeAlgorithm} algorithm
+   * @param {string} providerUri
+   * @param {Web3} web3
+   * @param {any} fetchMethod
+   * @param {ComputeOutput} output
+   * @return {Promise<ComputeJob | ComputeJob[]>}
+   */
+  public async computeStart(
+    did: string,
+    consumerAddress: string,
+    algorithm: ComputeAlgorithm,
+    providerUri: string,
     web3: Web3,
+    fetchMethod: any,
+    output?: ComputeOutput
+  ): Promise<ComputeJob | ComputeJob[]> {
+    const providerEndpoints = await this.getEndpoints(providerUri, fetchMethod)
+    const serviceEndpoints = await this.getServiceEndpoints(
+      providerUri,
+      providerEndpoints
+    )
+    const computeStartUrl = this.getEndpointURL(serviceEndpoints, 'computeStart')
+      ? this.getEndpointURL(serviceEndpoints, 'computeStart').urlPath
+      : null
+
+    const nonce = await this.getNonce(
+      providerUri,
+      consumerAddress,
+      fetchMethod,
+      providerEndpoints,
+      serviceEndpoints
+    )
+
+    let signatureMessage = consumerAddress
+    signatureMessage += (did && `${noZeroX(did)}`) || ''
+    signatureMessage += nonce
+    const signature = await this.createHashSignature(
+      web3,
+      consumerAddress,
+      signatureMessage
+    )
+
+    const payload = Object()
+    payload.consumerAddress = consumerAddress
+    payload.signature = signature
+    payload.algorithmDid = algorithm.did
+    payload.algorithmMeta = algorithm.meta
+    payload.algorithmServiceId = algorithm.serviceIndex
+    if (output) payload.output = output
+
+    if (!computeStartUrl) return null
+    try {
+      const response = await fetchMethod(computeStartUrl, JSON.stringify(payload))
+      if (response?.ok) {
+        const params = await response.json()
+        return params
+      }
+      console.error('Compute start failed:', response.status, response.statusText)
+      LoggerInstance.error('Payload was:', payload)
+      return null
+    } catch (e) {
+      LoggerInstance.error('Compute start failed:')
+      LoggerInstance.error(e)
+      LoggerInstance.error('Payload was:', payload)
+      return null
+    }
+  }
+
+  /** Instruct the provider to Stop the execution of a to stop a compute job.
+   * @param {string} did
+   * @param {string} consumerAddress
+   * @param {string} jobId
+   * @param {string} providerUri
+   * @param {Web3} web3
+   * @param {any} fetchMethod
+   * @return {Promise<ComputeJob | ComputeJob[]>}
+   */
+  public async computeStop(
+    did: string,
+    consumerAddress: string,
+    jobId: string,
+    providerUri: string,
+    web3: Web3,
+    fetchMethod: any
+  ): Promise<ComputeJob | ComputeJob[]> {
+    const providerEndpoints = await this.getEndpoints(providerUri, fetchMethod)
+    const serviceEndpoints = await this.getServiceEndpoints(
+      providerUri,
+      providerEndpoints
+    )
+    const computeStopUrl = this.getEndpointURL(serviceEndpoints, 'computeStop')
+      ? this.getEndpointURL(serviceEndpoints, 'computeStop').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
+    )
+
+    const payload = Object()
+    payload.signature = signature
+    payload.documentId = noZeroX(did)
+    payload.consumerAddress = consumerAddress
+    if (jobId) payload.jobId = jobId
+
+    if (!computeStopUrl) return null
+    try {
+      const response = await fetchMethod(computeStopUrl, JSON.stringify(payload))
+      if (response?.ok) {
+        const params = await response.json()
+        return params
+      }
+      LoggerInstance.error('Compute stop failed:', response.status, response.statusText)
+      LoggerInstance.error('Payload was:', payload)
+      return null
+    } catch (e) {
+      LoggerInstance.error('Compute stop failed:')
+      LoggerInstance.error(e)
+      LoggerInstance.error('Payload was:', payload)
+      return null
+    }
+  }
+
+  /** Get status for a specific jobId/documentId/owner.
+   * @param {string} did
+   * @param {string} consumerAddress
+   * @param {string} providerUri
+   * @param {Web3} web3
+   * @param {any} fetchMethod
+   * @param {string} jobId
+   * @return {Promise<ComputeJob | ComputeJob[]>}
+   */
+  public async computeStatus(
+    did: string,
+    consumerAddress: string,
+    providerUri: string,
+    web3: Web3,
+    fetchMethod: any,
+    jobId?: string
+  ): Promise<ComputeJob | ComputeJob[]> {
+    const providerEndpoints = await this.getEndpoints(providerUri, fetchMethod)
+    const serviceEndpoints = await this.getServiceEndpoints(
+      providerUri,
+      providerEndpoints
+    )
+    const computeStatusUrl = this.getEndpointURL(serviceEndpoints, 'computeStatus')
+      ? 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)
+      if (response?.ok) {
+        const params = await response.json()
+        return params
+      }
+      LoggerInstance.error(
+        'Get compute status failed:',
+        response.status,
+        response.statusText
+      )
+      return null
+    } catch (e) {
+      LoggerInstance.error('Get compute status failed')
+      LoggerInstance.error(e)
+      return null
+    }
+  }
+
+  /** Get status for a specific jobId/documentId/owner.
+   * @param {string} jobId
+   * @param {number} index
+   * @param {string} providerUri
+   * @param {string} destination
+   * @param {Web3} web3
+   * @param {any} fetchMethod
+   * @return {Promise<ComputeJob | ComputeJob[]>}
+   */
+  public async computeResult(
+    jobId: string,
+    index: number,
+    destination: string,
     accountId: string,
-    agreementId: string
-  ): Promise<string> {
-    const signature = await signText(web3, noZeroX(agreementId), accountId)
-    return signature
+    providerUri: string,
+    web3: Web3,
+    fetchMethod: any
+  ): Promise<any> {
+    const providerEndpoints = await this.getEndpoints(providerUri, fetchMethod)
+    const serviceEndpoints = await this.getServiceEndpoints(
+      providerUri,
+      providerEndpoints
+    )
+    const computeResultUrl = this.getEndpointURL(serviceEndpoints, 'computeResult')
+      ? this.getEndpointURL(serviceEndpoints, 'computeResult').urlPath
+      : null
+
+    const nonce = await this.getNonce(
+      providerUri,
+      accountId,
+      fetchMethod,
+      providerEndpoints,
+      serviceEndpoints
+    )
+
+    let signatureMessage = accountId
+    signatureMessage += jobId
+    signatureMessage += String(index)
+    signatureMessage += nonce
+    const signature = await this.createHashSignature(web3, accountId, signatureMessage)
+
+    let consumeUrl = computeResultUrl
+    consumeUrl += `?consumerAddress=${accountId}`
+    consumeUrl += `&jobId=${jobId}`
+    consumeUrl += `&index=${String(index)}`
+    consumeUrl += (signature && `&signature=${signature}`) || ''
+
+    if (!computeResultUrl) return null
+    try {
+      !destination
+        ? await fetchMethod.downloadFileBrowser(consumeUrl)
+        : await fetchMethod.downloadFile(consumeUrl, destination, index)
+    } catch (e) {
+      LoggerInstance.error('Error getting job result')
+      LoggerInstance.error(e)
+      throw e
+    }
+    return destination
+  }
+
+  /** Deletes a compute job.
+   * @param {string} did
+   * @param {string} consumerAddress
+   * @param {string} jobId
+   * @param {string} providerUri
+   * @param {Web3} web3
+   * @param {any} fetchMethod
+   * @return {Promise<ComputeJob | ComputeJob[]>}
+   */
+  public async computeDelete(
+    did: string,
+    consumerAddress: string,
+    jobId: string,
+    providerUri: string,
+    web3: Web3,
+    fetchMethod: any
+  ): Promise<ComputeJob | ComputeJob[]> {
+    const providerEndpoints = await this.getEndpoints(providerUri, fetchMethod)
+    const serviceEndpoints = await this.getServiceEndpoints(
+      providerUri,
+      providerEndpoints
+    )
+    const computeDeleteUrl = this.getEndpointURL(serviceEndpoints, 'computeDelete')
+      ? this.getEndpointURL(serviceEndpoints, 'computeDelete').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
+    )
+
+    const payload = Object()
+    payload.documentId = noZeroX(did)
+    payload.consumerAddress = consumerAddress
+    payload.jobId = jobId
+    if (signature) payload.signature = signature
+
+    if (!computeDeleteUrl) return null
+    try {
+      const response = await fetchMethod(computeDeleteUrl, JSON.stringify(payload))
+      if (response?.ok) {
+        const params = await response.json()
+        return params
+      }
+      LoggerInstance.error(
+        'Delete compute job failed:',
+        response.status,
+        response.statusText
+      )
+      LoggerInstance.error('Payload was:', payload)
+      return null
+    } catch (e) {
+      LoggerInstance.error('Delete compute job failed:')
+      LoggerInstance.error(e)
+      LoggerInstance.error('Payload was:', payload)
+      return null
+    }
   }
 
   /** Check for a valid provider at URL

src/utils/SignatureUtils.ts

@@ -26,3 +26,30 @@ export async function signText(
     }
   }
 }
+
+export async function signWithHash(
+  web3: Web3,
+  text: string,
+  publicKey: string,
+  password?: string
+): Promise<string> {
+  const hash = web3.utils.utf8ToHex(text)
+  const isMetaMask =
+    web3 && web3.currentProvider && (web3.currentProvider as any).isMetaMask
+  try {
+    return await web3.eth.personal.sign(hash, publicKey, password)
+  } catch (e) {
+    if (isMetaMask) {
+      throw e
+    }
+    LoggerInstance.warn('Error on personal sign.')
+    LoggerInstance.warn(e)
+    try {
+      return await web3.eth.sign(hash, publicKey)
+    } catch (e2) {
+      LoggerInstance.error('Error on sign.')
+      LoggerInstance.error(e2)
+      throw new Error('Error executing personal sign')
+    }
+  }
+}