feat(storage): add support for peerid tracking on offer

README.md

@@ -64,6 +64,7 @@ It has following schema:
 [
   {
     "address": "string", // Hash, serves as ID
+    "peerId": "string?", // Optional PeerId of the Provider's node
     "totalCapacity": "number",
     "utilizedCapacity": "number",
     "availableCapacity": "number",

src/services/storage/handlers/agreement.ts

@@ -1,31 +1,24 @@
 import { EventData } from 'web3-eth-contract'
-import { hexToAscii, soliditySha3 } from 'web3-utils'
+import { soliditySha3 } from 'web3-utils'
+import { Eth } from 'web3-eth'
 
 import { loggingFactory } from '../../../logger'
 import { Handler } from '../../../definitions'
 import { EventError } from '../../../errors'
+import { decodeByteArray } from '../../../utils'
+import { getBlockDate } from '../../../blockchain/utils'
 
 import Agreement from '../models/agreement.model'
 import BillingPlan from '../models/price.model'
-import { getBlockDate } from '../../../blockchain/utils'
-import { Eth } from 'web3-eth'
 import { StorageServices } from '../index'
 
 const logger = loggingFactory('storage:handler:request')
 
-function decodeDataReference (fileReference: string[]): string {
-  return fileReference
-    .map(hexToAscii)
-    .join('')
-    .trim()
-    .replace(/\0/g, '') // Remove null-characters
-}
-
 const handlers = {
   async NewAgreement (event: EventData, { agreementService }: StorageServices, eth: Eth): Promise<void> {
     const { provider: offerId, billingPeriod: period } = event.returnValues
     const id = soliditySha3(event.returnValues.agreementCreator, ...event.returnValues.dataReference)
-    const dataReference = decodeDataReference(event.returnValues.dataReference)
+    const dataReference = decodeByteArray(event.returnValues.dataReference)
 
     const plan = await BillingPlan.findOne({ where: { offerId, period } })
 

src/services/storage/handlers/offer.ts

@@ -4,6 +4,7 @@ import { EventData } from 'web3-eth-contract'
 import { loggingFactory } from '../../../logger'
 import { Handler } from '../../../definitions'
 import { StorageServices } from '../index'
+import { decodeByteArray } from '../../../utils'
 
 const logger = loggingFactory('storage:handler:offer')
 
@@ -38,10 +39,24 @@ const handler: Handler<StorageServices> = {
         offer.totalCapacity = event.returnValues.capacity
         logger.info(`Updating capacity ${offer.totalCapacity} (ID: ${offer.address})`)
         break
-      case 'MessageEmitted':
-        // TODO: Add Message support
-        logger.info(`Message ${event.returnValues.message} (ID: ${offer.address})`)
+      case 'MessageEmitted': {
+        const msg = event.returnValues.message
+
+        if (!msg || msg.length === 0) {
+          break
+        }
+
+        const [firstMsg, ...restMsg] = msg[0].replace('0x', '')
+        const flag = firstMsg.substring(0, 2)
+
+        if (flag === '01') { // PeerId definition
+          offer.peerId = decodeByteArray([firstMsg.substring(2), ...restMsg])
+          logger.info(`PeerId ${offer.peerId} defined (ID: ${offer.address})`)
+        } else {
+          logger.error(`Unknown message flag ${flag}!`)
+        }
         break
+      }
       case 'BillingPlanSet':
         await updatePrices(offer, event.returnValues.period, event.returnValues.price)
         break

src/services/storage/models/offer.model.ts

@@ -24,6 +24,9 @@ export default class Offer extends Model {
   @Column
   totalCapacity!: number
 
+  @Column
+  peerId!: string
+
   @HasMany(() => BillingPlan)
   plans!: BillingPlan[]
 

src/utils.ts

@@ -3,13 +3,14 @@ import { Input, OutputFlags } from '@oclif/parser'
 import { readFile as readFileCb } from 'fs'
 import { promisify } from 'util'
 import config from 'config'
+import { hexToAscii } from 'web3-utils'
 
 import { Application, Config, isSupportedServices, Logger, SupportedServices } from './definitions'
 
 const readFile = promisify(readFileCb)
 
 /**
- * Function that will split array into two groups based on callback
+ * Function that will split array into two groups based on callback that returns Promise.
  *
  * @param arr
  * @param callback
@@ -22,16 +23,60 @@ export async function asyncSplit<T> (arr: T[], callback: (elem: T) => Promise<bo
   }, [[], []])
 }
 
+/**
+ * Utility function that will split array into two groups based on sync callback.
+ * @param array
+ * @param isValid
+ * @return [success, failure] array where first are positives based on callback and second are negatives
+ */
 export function split<T> (array: T[], isValid: (elem: T) => boolean): [T[], T[]] {
   return array.reduce<[T[], T[]]>(([pass, fail], elem) => {
     return isValid(elem) ? [[...pass, elem], fail] : [pass, [...fail, elem]]
   }, [[], []])
 }
 
+/**
+ * Utility function for decoding Solidity's byte32 array.
+ * @param fileReference
+ */
+export function decodeByteArray (fileReference: string[]): string {
+  return fileReference
+    .map(hexToAscii)
+    .join('')
+    .trim()
+    .replace(/\0/g, '') // Remove null-characters
+}
+
+/**
+ * Utility function that capitalize first letter of given string
+ * @param value
+ */
 export function capitalizeFirstLetter (value: string): string {
   return value.charAt(0).toUpperCase() + value.slice(1)
 }
 
+/**
+ * Takes Sets A and B and create a difference of those, which results in a subset of A, where
+ * elements from set B are removed.
+ * @param setA
+ * @param setB
+ */
+export function setDifference<T> (setA: Set<T>, setB: Set<T>): Set<T> {
+  const _difference = new Set(setA)
+  for (const elem of setB) {
+    _difference.delete(elem)
+  }
+  return _difference
+}
+
+/**
+ * Function mainly for CLI that validates the given list of string if they are valid service's names
+ * or if it is "all" (in case that is allowed).
+ *
+ * It returns the list of services. In case of "all" then it is expanded to list of all service's names.
+ * @param args
+ * @param onlyEnabledForAll
+ */
 export function validateServices (args: string[], onlyEnabledForAll = false): SupportedServices[] {
   if (args.length === 0) {
     throw new Error('You have to specify a service!')
@@ -55,12 +100,23 @@ export function validateServices (args: string[], onlyEnabledForAll = false): Su
   return args as SupportedServices[]
 }
 
+/**
+ * General handler closure function mainly for Event Emitters, which in case of rejected promise logs the rejection
+ * using given logger.
+ *
+ * @param fn
+ * @param logger
+ */
 export function errorHandler (fn: (...args: any[]) => Promise<void>, logger: Logger): (...args: any[]) => Promise<void> {
   return (...args) => {
     return fn(...args).catch(err => logger.error(err))
   }
 }
 
+/**
+ * Helper function which awaits on all the initializations Promises that are set on the `app` object.
+ * @param app
+ */
 export async function waitForReadyApp (app: Application): Promise<void> {
   await app.get('storeInit')
   await app.get('sequelizeSync')
@@ -120,11 +176,3 @@ export abstract class BaseCLICommand extends Command {
     return Promise.resolve()
   }
 }
-
-export function setDifference<T> (setA: Set<T>, setB: Set<T>): Set<T> {
-  const _difference = new Set(setA)
-  for (const elem of setB) {
-    _difference.delete(elem)
-  }
-  return _difference
-}