Feature/dht search (#967)

Author: alexcos20

docs/API.md

@@ -404,19 +404,58 @@ returns list of logs
 
 ---
 
-## Advertise Did
+## Get providers for a string
 
-### `HTTP` GET /advertiseDid/?did=did:op:123"
+### `HTTP` GET /getProvidersForString/?input=did:op:123"
 
 #### Description
 
-returns empty if advertising did around peers was successful
+returns list of nodes providing the specific element(s) (dids, c2d resources, etc)
 
 #### Query Parameters
 
-| name | type   | required | description        |
-| ---- | ------ | -------- | ------------------ |
-| did  | string | v        | document id or did |
+| name  | type   | required | description            |
+| ----- | ------ | -------- | ---------------------- |
+| input | string | v        | did, c2d resource, etc |
+
+## Get providers for a list of strings
+
+### `HTTP` POST /getProvidersForStrings?timeout=10"
+
+#### Description
+
+returns list of nodes providing all specific elements.
+
+#### Query Parameters
+
+| name    | type   | required | description            |
+| ------- | ------ | -------- | ---------------------- |
+| timeout | string | optional | timeout in miliseconds |
+
+#### Request
+
+```json
+["{\"c2d\":{\"free\":false,\"disk\":1}}", "{\"c2d\":{\"free\":false,\"cpu\":1}}"]
+```
+
+#### Response
+
+```json
+[
+  {
+    "id": "16Uiu2HAmENNgCY1QAdQrPxipgUCQjyookUgpnbgXua4ZMju4Rkou",
+    "multiaddrs": [
+      "/ip4/10.255.255.254/tcp/41015/ws",
+      "/ip4/10.255.255.254/tcp/41347",
+      "/ip4/127.0.0.1/tcp/41015/ws",
+      "/ip4/127.0.0.1/tcp/41347",
+      "/ip4/172.27.58.101/tcp/41015/ws",
+      "/ip4/172.27.58.101/tcp/41347",
+      "/ip6/::1/tcp/37527"
+    ]
+  }
+]
+```
 
 ---
 
@@ -1404,7 +1443,7 @@ returns job result
 | jobId           | string | v        | jobId address to use as filter                                 |
 | signature       | string | v        | signature (consumerAddress + jobId + index.toString() + nonce) |
 | nonce           | string | v        | nonce for the request                                          |
-| index           | number | v        | index of result  (0 for main result, 1 for logs)               |
+| index           | number | v        | index of result (0 for main result, 1 for logs)                |
 
 #### Response
 

src/@types/C2D/C2D.ts

@@ -51,7 +51,7 @@ export interface ComputeResource {
   id: ComputeResourceType
   description?: string
   type?: string
-  kind?: string
+  kind?: string // discreet, named, etc
   total: number // total number of specific resource
   min: number // min number of resource needed for a job
   max: number // max number of resource for a job

src/components/P2P/index.ts

@@ -1,5 +1,5 @@
 // import diff from 'hyperdiff'
-import { P2PCommandResponse, TypesenseSearchResponse } from '../../@types/index'
+import { P2PCommandResponse } from '../../@types/index'
 import EventEmitter from 'node:events'
 import clone from 'lodash.clonedeep'
 
@@ -61,10 +61,6 @@ type DDOCache = {
   dht: Map<string, FindDDOResponse>
 }
 
-// republish any ddos we are providing to the network every 4 hours
-// (we can put smaller interval for testing purposes)
-const REPUBLISH_INTERVAL_HOURS = 1000 * 60 * 60 * 4 // 4 hours
-
 let index = 0
 
 export class OceanP2P extends EventEmitter {
@@ -127,15 +123,13 @@ export class OceanP2P extends EventEmitter {
     this._connections = {}
     this._protocol = '/ocean/nodes/1.0.0'
 
-    // this._interval = setInterval(this._pollPeers.bind(this), this._options.pollInterval)
+    this._interval = setInterval(this._flushAdvertiseQueue.bind(this), 60 * 1000) // every 60 seconds
 
     // only enable handling of commands if not bootstrap node
     if (!this._config.isBootstrap) {
       this._libp2p.handle(this._protocol, handleProtocolCommands.bind(this))
     }
 
-    setInterval(this.republishStoredDDOS.bind(this), REPUBLISH_INTERVAL_HOURS)
-
     this._idx = index++
 
     this._analyzeRemoteResponse = new Transform({
@@ -146,7 +140,7 @@ export class OceanP2P extends EventEmitter {
     // listen for indexer events and advertise did
     INDEXER_DDO_EVENT_EMITTER.addListener(EVENTS.METADATA_CREATED, (did) => {
       P2P_LOGGER.info(`Listened "${EVENTS.METADATA_CREATED}"`)
-      this.advertiseDid(did)
+      this.advertiseString(did)
     })
   }
 
@@ -755,17 +749,32 @@ export class OceanP2P extends EventEmitter {
     }
   }
 
-  async advertiseDid(did: string) {
-    P2P_LOGGER.logMessage('Advertising ' + did, true)
+  async _flushAdvertiseQueue() {
+    if (this._pendingAdvertise.length > 0) {
+      P2P_LOGGER.debug(
+        `Flushing advertise queue with ${this._pendingAdvertise.length} items`
+      )
+      const list = JSON.parse(JSON.stringify(this._pendingAdvertise))
+      for (const did of list) {
+        this._pendingAdvertise = this._pendingAdvertise.filter((item) => item !== did)
+
+        await this.advertiseString(did)
+      }
+      // this._pendingAdvertise = []
+    }
+  }
+
+  async advertiseString(did: string) {
     try {
+      const cid = await cidFromRawString(did)
+      P2P_LOGGER.debug('Advertising  "' + did + `" as CID:` + cid)
       const x = (await this.getAllOceanPeers()).length
       if (x > 0) {
-        const cid = await cidFromRawString(did)
         const multiAddrs = this._libp2p.components.addressManager.getAddresses()
         // console.log('multiaddrs: ', multiAddrs)
-        await this._libp2p.contentRouting.provide(cid, multiAddrs)
+        this._libp2p.contentRouting.provide(cid, multiAddrs)
       } else {
-        P2P_LOGGER.verbose(
+        P2P_LOGGER.debug(
           'Could not find any Ocean peers. Nobody is listening at the moment, skipping...'
         )
         // save it for retry later
@@ -779,67 +788,50 @@ export class OceanP2P extends EventEmitter {
     }
   }
 
-  async getProvidersForDid(did: string) {
-    P2P_LOGGER.logMessage('Fetching providers for ' + did, true)
-    const cid = await cidFromRawString(did)
+  getCommonPeers(
+    rets: Array<Array<{ id: string; multiaddrs: any[] }>>
+  ): Array<{ id: string; multiaddrs: any[] }> {
+    return rets.reduce(
+      (acc, curr) =>
+        acc.filter((item) => curr.some((el) => el.id.toString() === item.id.toString())),
+      rets[0] // Initialize with first subarray
+    )
+  }
+
+  async getProvidersForStrings(
+    input: string[],
+    timeout?: number
+  ): Promise<Array<{ id: string; multiaddrs: any[] }>> {
+    const rets = await Promise.all(
+      input.map(async (x) => {
+        const providers = await this.getProvidersForString(x, timeout)
+        return providers && providers.length > 0 ? providers : [] // Keep only valid results
+      })
+    )
+    return this.getCommonPeers(rets)
+  }
+
+  async getProvidersForString(
+    input: string,
+    timeout?: number
+  ): Promise<Array<{ id: string; multiaddrs: any[] }>> {
+    P2P_LOGGER.logMessage('Fetching providers for ' + input, true)
+    const cid = await cidFromRawString(input)
     const peersFound = []
     try {
       const f = await this._libp2p.contentRouting.findProviders(cid, {
-        queryFuncTimeout: 20000 // 20 seconds
+        queryFuncTimeout: timeout || 20000 // 20 seconds
         // on timeout the query ends with an abort signal => CodeError: Query aborted
       })
       for await (const value of f) {
         peersFound.push(value)
       }
     } catch (e) {
-      P2P_LOGGER.error('getProvidersForDid()' + e.message)
+      P2P_LOGGER.error('getProvidersForString()' + e.message)
     }
     return peersFound
   }
 
-  // republish the ddos we have
-  // related: https://github.com/libp2p/go-libp2p-kad-dht/issues/323
-  async republishStoredDDOS() {
-    try {
-      if (!this.db || !this.db.ddo) {
-        P2P_LOGGER.logMessage(
-          `republishStoredDDOS() attempt aborted because there is no database!`,
-          true
-        )
-        return
-      }
-      const db = this.db.ddo
-      const searchParameters = {
-        q: '*'
-      }
-
-      const result: TypesenseSearchResponse[] = await db.search(searchParameters)
-      if (result && result.length > 0 && result[0].found) {
-        P2P_LOGGER.logMessage(`Will republish cid for ${result[0].found} documents`, true)
-        result[0].hits.forEach((hit: any) => {
-          const ddo = hit.document
-          this.advertiseDid(ddo.id)
-          // populate hash table if not exists
-          // (even if no peers are listening, it still goes to the pending publish table)
-          if (!this._ddoDHT.dht.has(ddo.id)) {
-            this.cacheDDO(ddo)
-          }
-          // todo check stuff like purgatory
-        })
-        // update time
-        this._ddoDHT.updated = new Date().getTime()
-      } else {
-        P2P_LOGGER.logMessage('There is nothing to republish, skipping...', true)
-      }
-    } catch (err) {
-      P2P_LOGGER.log(
-        LOG_LEVELS_STR.LEVEL_ERROR,
-        `Caught "${err.message}" on republishStoredDDOS()`,
-        true
-      )
-    }
-  }
-
   // cache a ddos object
   cacheDDO(ddo: any) {
     this._ddoDHT.dht.set(ddo.id, {
@@ -848,6 +840,7 @@ export class OceanP2P extends EventEmitter {
       lastUpdateTime: ddo.metadata.updated,
       provider: this.getPeerId()
     })
+    this._ddoDHT.updated = new Date().getTime()
   }
 
   /**
@@ -893,7 +886,7 @@ export class OceanP2P extends EventEmitter {
         // if already added before, create() will return null, but still advertise it
         try {
           await db.create(ddo)
-          await this.advertiseDid(ddo.id)
+          await this.advertiseString(ddo.id)
           // populate hash table
           this.cacheDDO(ddo)
           count++

src/components/c2d/compute_engine_docker.ts

@@ -172,12 +172,14 @@ export class C2DEngineDocker extends C2DEngine {
     this.envs[0].resources = []
     this.envs[0].resources.push({
       id: 'cpu',
+      type: 'cpu',
       total: sysinfo.NCPU,
       max: sysinfo.NCPU,
       min: 1
     })
     this.envs[0].resources.push({
       id: 'ram',
+      type: 'ram',
       total: sysinfo.MemTotal,
       max: sysinfo.MemTotal,
       min: 1e9

src/components/core/handler/ddoHandler.ts

@@ -616,7 +616,7 @@ export class FindDdoHandler extends CommandHandler {
       }, 1000 * MAX_RESPONSE_WAIT_TIME_SECONDS)
 
       // check other providers for this ddo
-      const providers = await p2pNode.getProvidersForDid(task.id)
+      const providers = await p2pNode.getProvidersForString(task.id)
       // check if includes self and exclude from check list
       if (providers.length > 0) {
         // exclude this node from the providers list if present

src/components/httpRoutes/dids.ts

@@ -2,41 +2,56 @@ import express, { Request, Response } from 'express'
 import { sendMissingP2PResponse } from './index.js'
 import { hasP2PInterface } from '../../utils/config.js'
 
-export const advertiseDidRoute = express.Router()
-
-advertiseDidRoute.post(
-  '/advertiseDid',
+export const getProvidersForStringRoute = express.Router()
+getProvidersForStringRoute.get(
+  '/getProvidersForString',
   express.urlencoded({ extended: true, type: '*/*' }),
   async (req: Request, res: Response): Promise<void> => {
-    if (!req.query.did) {
+    if (!req.query.input) {
       res.sendStatus(400)
       return
     }
     if (hasP2PInterface) {
-      await req.oceanNode.getP2PNode().advertiseDid(req.query.did as string)
-      res.sendStatus(200)
+      const providers = await req.oceanNode
+        .getP2PNode()
+        .getProvidersForString(req.query.input as string)
+      res.json(providers)
     } else {
       sendMissingP2PResponse(res)
     }
   }
 )
 
-export const getProvidersForDidRoute = express.Router()
-getProvidersForDidRoute.get(
-  '/getProvidersForDid',
-  express.urlencoded({ extended: true, type: '*/*' }),
-  async (req: Request, res: Response): Promise<void> => {
-    if (!req.query.did) {
-      res.sendStatus(400)
-      return
-    }
-    if (hasP2PInterface) {
-      const providers = await req.oceanNode
-        .getP2PNode()
-        .getProvidersForDid(req.query.did as string)
-      res.json(providers)
-    } else {
-      sendMissingP2PResponse(res)
+export const getProvidersForStringsRoute = express.Router()
+getProvidersForStringsRoute.post(
+  '/getProvidersForStrings',
+  express.json(),
+  async (req, res) => {
+    try {
+      if (!req.body) {
+        res.status(400).send('Missing array of strings in request body.')
+        return
+      }
+      // const body = JSON.parse(req.body)
+      if (
+        Array.isArray(req.body) &&
+        req.body.every((item: unknown) => typeof item === 'string')
+      ) {
+        const timeout =
+          typeof req.query?.timeout === 'string'
+            ? parseInt(req.query.timeout, 10)
+            : undefined
+        const providers = await req.oceanNode
+          .getP2PNode()
+          .getProvidersForStrings(req.body, timeout)
+
+        res.json(providers)
+      } else {
+        res.status(400).send('Expected an array of strings.')
+      }
+    } catch (error) {
+      console.error('Error processing request:', error)
+      res.status(400).send(error)
     }
   }
 )