publish & consume flow working (#1192)

* simple publish & consume flow working

src/aquarius/Aquarius.ts

@@ -1,5 +1,5 @@
 import { LoggerInstance } from '../utils'
-import { DDO } from '../@types/'
+import { Asset, DDO } from '../@types/'
 
 export class Aquarius {
   public aquariusURL
@@ -19,11 +19,9 @@ export class Aquarius {
   public async resolve(did: string, fetchMethod: any): Promise<DDO> {
     const path = this.aquariusURL + '/api/aquarius/assets/ddo/' + did
     try {
-      console.log(path)
+      const response = await fetchMethod('GET', path)
-      const response = await fetchMethod(path)
       if (response.ok) {
         const raw = await response.json()
-        console.log(raw)
         return raw as DDO
       } else {
         throw new Error('HTTP request failed with status ' + response.status)
@@ -50,18 +48,18 @@ export class Aquarius {
    * @param {string} fetchMethod fetch client instance
    * @return {Promise<DDO>} DDO of the asset.
    */
-  public async waitForAqua(fetchMethod: any, did: string, txid?: string): Promise<DDO> {
+  public async waitForAqua(fetchMethod: any, did: string, txid?: string): Promise<Asset> {
     let tries = 0
     do {
       try {
         const path = this.aquariusURL + '/api/aquarius/assets/ddo/' + did
-        const response = await fetchMethod(path)
+        const response = await fetchMethod('GET', path)
         if (response.ok) {
           const ddo = await response.json()
           if (txid) {
             // check tx
-            if (ddo.event && ddo.event.txid === txid) return ddo as DDO
+            if (ddo.event && ddo.event.txid === txid) return ddo as Asset
-          } else return ddo as DDO
+          } else return ddo as Asset
         }
       } catch (e) {
         // do nothing

src/provider/Provider.ts

@@ -132,7 +132,7 @@ export class Provider {
 
     if (!path) return null
     try {
-      const response = await postMethod(path, decodeURI(JSON.stringify(data)), {
+      const response = await postMethod('POST', path, decodeURI(JSON.stringify(data)), {
         'Content-Type': 'application/octet-stream'
       })
       return response
@@ -201,7 +201,7 @@ export class Provider {
       : null
     if (!path) return null
     try {
-      const response = await fetchMethod(path, JSON.stringify(args))
+      const response = await fetchMethod('POST', path, JSON.stringify(args))
       const results: FileMetadata[] = await response.json()
       for (const result of results) {
         files.push(result)
@@ -223,9 +223,9 @@ export class Provider {
    * @return {Promise<ProviderInitialize>} ProviderInitialize data
    */
   public async initialize(
-    asset: Asset,
+    did: string,
-    serviceIndex: number,
+    serviceId: string,
-    serviceType: string,
+    fileIndex: number,
     consumerAddress: string,
     providerUri: string,
     getMethod: any,
@@ -241,43 +241,41 @@ export class Provider {
       : null
 
     if (!initializeUrl) return null
-    initializeUrl += `?documentId=${asset.id}`
+    initializeUrl += `?documentId=${did}`
-    initializeUrl += `&serviceId=${serviceIndex}`
+    initializeUrl += `&serviceId=${serviceId}`
-    initializeUrl += `&serviceType=${serviceType}`
+    initializeUrl += `&fileIndex=${fileIndex}`
-    initializeUrl += `&dataToken=${asset.datatokens[0]}` // to check later
     initializeUrl += `&consumerAddress=${consumerAddress}`
     if (userCustomParameters)
       initializeUrl += '&userdata=' + encodeURI(JSON.stringify(userCustomParameters))
     try {
-      const response = await getMethod(initializeUrl)
+      const response = await getMethod('GET', initializeUrl)
-      return (await response.json()) as ProviderInitialize
+      const results: ProviderInitialize = await response.json()
+      return results
     } catch (e) {
       LoggerInstance.error(e)
       throw new Error('Asset URL not found or not available.')
     }
   }
 
-  /** Allows download of asset data file.
+  /** Gets fully signed URL for download
    * @param {string} did
-   * @param {string} destination
    * @param {string} accountId
-   * @param {FileMetadata[]} files
+   * @param {string} serviceId
-   * @param {-1} index
+   * @param {number} fileIndex
    * @param {string} providerUri
    * @param {Web3} web3
    * @param {any} fetchMethod
    * @param {UserCustomParameters} userCustomParameters
-   * @return {Promise<any>}
+   * @return {Promise<string>}
    */
-  public async download(
+  public async getDownloadUrl(
     did: string,
-    destination: string,
     accountId: string,
-    files: FileMetadata[],
+    serviceId: string,
-    index = -1,
+    fileIndex: number,
+    transferTxId: string,
     providerUri: string,
     web3: Web3,
-    fetchMethod: any,
     userCustomParameters?: UserCustomParameters
   ): Promise<any> {
     const providerEndpoints = await this.getEndpoints(providerUri)
@@ -288,39 +286,21 @@ export class Provider {
     const downloadUrl = this.getEndpointURL(serviceEndpoints, 'download')
       ? this.getEndpointURL(serviceEndpoints, 'download').urlPath
       : null
-
+    if (!downloadUrl) return null
-    const nonce = await this.getNonce(
+    const nonce = Date.now()
-      providerUri,
-      accountId,
-      fetchMethod,
-      providerEndpoints,
-      serviceEndpoints
-    )
     const signature = await this.createSignature(web3, accountId, did + nonce)
 
-    if (!downloadUrl) return null
-    const filesPromises = files
-      .filter((_, i) => index === -1 || i === index)
-      .map(async ({ index: i, url: fileUrl }) => {
     let consumeUrl = downloadUrl
-        consumeUrl += `?index=${i}`
+    consumeUrl += `?fileIndex=${fileIndex}`
     consumeUrl += `&documentId=${did}`
+    consumeUrl += `&transferTxId=${transferTxId}`
+    consumeUrl += `&serviceId=${serviceId}`
     consumeUrl += `&consumerAddress=${accountId}`
-        consumeUrl += `&url=${fileUrl}`
+    consumeUrl += `&nonce=${nonce}`
     consumeUrl += `&signature=${signature}`
     if (userCustomParameters)
       consumeUrl += '&userdata=' + encodeURI(JSON.stringify(userCustomParameters))
-        try {
+    return consumeUrl
-          !destination
-            ? await fetchMethod.downloadFileBrowser(consumeUrl)
-            : await fetchMethod.downloadFile(consumeUrl, destination, i)
-        } catch (e) {
-          LoggerInstance.error('Error consuming assets', e)
-          throw e
-        }
-      })
-    await Promise.all(filesPromises)
-    return destination
   }
 
   /** Instruct the provider to start a compute job

src/utils/FetchHelper.ts

@@ -1,5 +1,7 @@
 import fetch from 'cross-fetch'
 import LoggerInstance from './Logger'
+import fs from 'fs'
+import save from 'save-file'
 
 export async function fetchData(url: string, opts: RequestInit): Promise<Response> {
   const result = await fetch(url, opts)
@@ -11,6 +13,37 @@ export async function fetchData(url: string, opts: RequestInit): Promise<Respons
   return result
 }
 
+export async function downloadFile(
+  url: string,
+  destination?: string,
+  index?: number
+): Promise<string> {
+  const response = await fetch(url)
+  if (!response.ok) {
+    throw new Error('Response error.')
+  }
+  let filename: string
+  try {
+    filename = response.headers
+      .get('content-disposition')
+      .match(/attachment;filename=(.+)/)[1]
+  } catch {
+    try {
+      filename = url.split('/').pop()
+    } catch {
+      filename = `file${index}`
+    }
+  }
+  if (destination) {
+    // eslint-disable-next-line no-async-promise-executor
+    fs.mkdirSync(destination, { recursive: true })
+    fs.writeFileSync(`${destination}/${filename}`, await response.text())
+    return destination
+  } else {
+    save(await response.arrayBuffer(), filename)
+  }
+}
+
 export async function getData(url: string): Promise<Response> {
   return fetch(url, {
     method: 'GET',
@@ -44,3 +77,17 @@ export async function postData(url: string, payload: BodyInit): Promise<Response
   }
   return postWithHeaders(url, payload, headers)
 }
+
+// simple fetch function used in tests
+export async function crossFetchGeneric(
+  method: string,
+  url: string,
+  body: string,
+  headers: any
+) {
+  return fetch(url, {
+    method: method,
+    body: body,
+    headers: headers
+  })
+}

test/integration/SimplePublishConsumeFlow.test.ts

@@ -3,6 +3,7 @@ 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'
@@ -11,6 +12,7 @@ import fetch from 'cross-fetch'
 import { 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(
@@ -55,7 +57,7 @@ const ddo = {
   services: [
     {
       id: 'notAnId',
-      type: 'download',
+      type: 'access',
       files: '',
       datatokenAddress: '0xa15024b732A8f2146423D14209eFd074e61964F3',
       serviceEndpoint: 'https://providerv4.rinkeby.oceanprotocol.com',
@@ -67,9 +69,11 @@ const ddo = {
 describe('Publish tests', async () => {
   it('should publish a dataset (create NFT + ERC20)', 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 accountId = accounts[0]
+    const publisherAccount = accounts[0]
+    const consumerAccount = accounts[1]
     const nftParams: NftCreateData = {
       name: 'testNFT',
       symbol: 'TST',
@@ -82,25 +86,22 @@ describe('Publish tests', async () => {
       feeAmount: '0',
       feeManager: '0x0000000000000000000000000000000000000000',
       feeToken: '0x0000000000000000000000000000000000000000',
-      minter: accountId,
+      minter: publisherAccount,
       mpFeeAddress: '0x0000000000000000000000000000000000000000'
     }
-    const result = await Factory.createNftWithErc(accountId, nftParams, erc20Params)
+    const result = await Factory.createNftWithErc(
+      publisherAccount,
+      nftParams,
+      erc20Params
+    )
     const erc721Address = result.events.NFTCreated.returnValues[0]
     const datatokenAddress = result.events.TokenCreated.returnValues[0]
 
     // create the files encrypted string
     let providerResponse = await ProviderInstance.encrypt(
-      ddo,
+      assetUrl,
       providerUrl,
-      (url: string, body: string, headers: any) => {
+      crossFetchGeneric
-        // replace with fetch
-        return fetch(url, {
-          method: 'POST',
-          body: body,
-          headers: headers
-        })
-      }
     )
     ddo.services[0].files = await providerResponse.text()
     ddo.services[0].datatokenAddress = datatokenAddress
@@ -110,23 +111,12 @@ describe('Publish tests', async () => {
     ddo.id =
       'did:op:' + SHA256(web3.utils.toChecksumAddress(erc721Address) + chain.toString(10))
 
-    providerResponse = await ProviderInstance.encrypt(
+    providerResponse = await ProviderInstance.encrypt(ddo, providerUrl, crossFetchGeneric)
-      ddo,
-      providerUrl,
-      (url: string, body: string, headers: any) => {
-        // replace with fetch
-        return fetch(url, {
-          method: 'POST',
-          body: body,
-          headers: headers
-        })
-      }
-    )
     const encryptedResponse = await providerResponse.text()
     const metadataHash = getHash(JSON.stringify(ddo))
     const res = await nft.setMetadata(
       erc721Address,
-      accountId,
+      publisherAccount,
       0,
       providerUrl,
       '',
@@ -134,13 +124,48 @@ describe('Publish tests', async () => {
       encryptedResponse,
       '0x' + metadataHash
     )
-    const resolvedDDO = await aquarius.waitForAqua((url: string, body: string) => {
+    const resolvedDDO = await aquarius.waitForAqua(crossFetchGeneric, ddo.id)
-      // replace with fetch
-      return fetch(url, {
-        method: 'GET',
-        body: body
-      })
-    }, ddo.id)
     assert(resolvedDDO, 'Cannot fetch DDO from Aquarius')
+    // mint 1 ERC20 and send it to the consumer
+    await datatoken.mint(datatokenAddress, publisherAccount, '1', consumerAccount)
+    // initialize provider
+    const initializeData = await ProviderInstance.initialize(
+      resolvedDDO.id,
+      resolvedDDO.services[0].id,
+      0,
+      consumerAccount,
+      providerUrl,
+      crossFetchGeneric
+    )
+    // make the payment
+    const txid = await datatoken.startOrder(
+      datatokenAddress,
+      consumerAccount,
+      consumerAccount,
+      0,
+      initializeData.providerFee.providerFeeAddress,
+      initializeData.providerFee.providerFeeToken,
+      initializeData.providerFee.providerFeeAmount,
+      initializeData.providerFee.v,
+      initializeData.providerFee.r,
+      initializeData.providerFee.s,
+      initializeData.providerFee.providerData
+    )
+    // get the url
+    const downloadURL = await ProviderInstance.getDownloadUrl(
+      ddo.id,
+      consumerAccount,
+      ddo.services[0].id,
+      0,
+      txid.transactionHash,
+      providerUrl,
+      web3
+    )
+    assert(downloadURL, 'Provider getDownloadUrl failed')
+    try {
+      await downloadFile(downloadURL, './tmpfile')
+    } catch (e) {
+      assert.fail('Download failed')
+    }
   })
 })