PoR (#914)

* feat: add fetcher feat to POR

* feat: add por run script

* chore: remove unused Error

* chore: remove unused Error

* feat: make separate api file for por

* feat: separate fetcher from por main file

* feat: store fetcherd data throught orakl-api

* feat: imrove por fetcher

* feat: make reporter for PoR

* fix: remove unnecessary try/catch wrapper

* fix: remove unnecessary try/catch wrapper

* feat: add heartbeat and threashold check to por submission

* feat: imrove logs

* feat: import aggregator-hash from variables

* fix: resolve lint errors

* feat: add new env variable

* feat: bump-up orakl-core  version from 0.4.0 to 0.5.0

* fix: remove hard value

* fix: deviation check

core/.env.example

@@ -17,3 +17,6 @@ HOST_SETTINGS_LOG_DIR=
 
 # Optional
 SLACK_WEBHOOK_URL=
+
+# PoR required variable
+POR_AGGREGATOR_HASH=

core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@bisonai/orakl-core",
-  "version": "0.4.0",
+  "version": "0.5.0",
   "type": "module",
   "description": "The Orakl Network Core",
   "files": [
@@ -44,6 +44,7 @@
     "start:reporter:vrf": "yarn start:reporter --reporter VRF",
     "start:reporter:request_response": "yarn start:reporter --reporter REQUEST_RESPONSE",
     "start:reporter:data_feed": "yarn start:reporter --reporter DATA_FEED",
+    "start:por": "node --no-warnings --experimental-specifier-resolution=node --experimental-json-modules dist/por/main.js",
     "start:reporter:data_feed_l2": "yarn start:reporter --reporter DATA_FEED_L2",
     "keygen": "node --no-warnings --experimental-specifier-resolution=node --experimental-json-modules dist/cli/vrf-keygen.js",
     "adapter-hash": "node --no-warnings --experimental-specifier-resolution=node --experimental-json-modules dist/cli/generate-adapter-hash.js",

core/src/errors.ts

@@ -16,7 +16,6 @@ export enum OraklErrorCode {
   GetListenerRequestFailed,
   GetReporterRequestFailed,
   GetVrfConfigRequestFailed,
-  IncompleteDataFeed,
   IndexOutOfBoundaries,
   InvalidAdapter,
   InvalidAggregator,
@@ -60,5 +59,6 @@ export enum OraklErrorCode {
   UnknownRequestResponseJob,
   MissingSignedRawTx,
   CaverTxTransactionFailed,
-  DelegatorServerIssue
+  DelegatorServerIssue,
+  FailedInsertData
 }

core/src/por/api.ts

@@ -0,0 +1,50 @@
+import axios from 'axios'
+import { OraklError, OraklErrorCode } from '../errors'
+import { CHAIN, ORAKL_NETWORK_API_URL } from '../settings'
+import { IAggregator } from '../types'
+import { buildUrl } from '../utils'
+import { IData } from './types'
+
+export async function loadAggregator({ aggregatorHash }: { aggregatorHash: string }) {
+  const chain = CHAIN
+  try {
+    const url = buildUrl(ORAKL_NETWORK_API_URL, `aggregator/${aggregatorHash}/${chain}`)
+    const aggregator: IAggregator = (await axios.get(url))?.data
+    return aggregator
+  } catch (e) {
+    throw new OraklError(OraklErrorCode.FailedToGetAggregator)
+  }
+}
+
+export async function insertData({
+  aggregatorId,
+  feedId,
+  value
+}: {
+  aggregatorId: bigint
+  feedId: bigint
+  value: number
+}) {
+  const timestamp = new Date(Date.now()).toString()
+  const data: IData[] = [
+    {
+      aggregatorId: aggregatorId.toString(),
+      feedId,
+      timestamp,
+      value
+    }
+  ]
+
+  try {
+    const url = buildUrl(ORAKL_NETWORK_API_URL, 'data')
+    const response = await axios.post(url, { data })
+
+    return {
+      status: response?.status,
+      statusText: response?.statusText,
+      data: response?.data
+    }
+  } catch (e) {
+    throw new OraklError(OraklErrorCode.FailedInsertData)
+  }
+}

core/src/por/errors.ts

@@ -0,0 +1,17 @@
+export class PorError extends Error {
+  constructor(public readonly code: PorErrorCode, message?: string, public readonly value?) {
+    super(message)
+    this.name = PorErrorCode[code]
+    this.value = value
+    Object.setPrototypeOf(this, new.target.prototype)
+  }
+}
+
+export enum PorErrorCode {
+  IndexOutOfBoundaries,
+  InvalidAdapter,
+  InvalidDataFeed,
+  InvalidDataFeedFormat,
+  InvalidReducer,
+  MissingKeyInJson
+}

core/src/por/fetcher.ts

@@ -0,0 +1,57 @@
+import axios from 'axios'
+import { IAggregator } from '../types'
+import { pipe } from '../utils'
+import { insertData, loadAggregator } from './api'
+import { PorError, PorErrorCode } from './errors'
+import { DATA_FEED_REDUCER_MAPPING } from './reducer'
+
+async function extractFeed(adapter) {
+  const feeds = adapter.feeds.map((f) => {
+    return {
+      id: f.id,
+      name: f.name,
+      url: f.definition.url,
+      headers: f.definition.headers,
+      method: f.definition.method,
+      reducers: f.definition.reducers
+    }
+  })
+  return feeds[0]
+}
+
+function checkDataFormat(data) {
+  if (!data) {
+    throw new PorError(PorErrorCode.InvalidDataFeed)
+  } else if (!Number.isInteger(data)) {
+    throw new PorError(PorErrorCode.InvalidDataFeedFormat)
+  }
+}
+
+function buildReducer(reducerMapping, reducers) {
+  return reducers.map((r) => {
+    const reducer = reducerMapping[r.function]
+    if (!reducer) {
+      throw new PorError(PorErrorCode.InvalidReducer)
+    }
+    return reducer(r?.args)
+  })
+}
+
+async function fetchData(feed) {
+  const rawDatum = await (await axios.get(feed.url)).data
+  const reducers = await buildReducer(DATA_FEED_REDUCER_MAPPING, feed.reducers)
+  const datum = pipe(...reducers)(rawDatum)
+  checkDataFormat(datum)
+  return datum
+}
+
+export async function fetchWithAggregator(aggregatorHash: string) {
+  const aggregator: IAggregator = await loadAggregator({ aggregatorHash })
+  const adapter = aggregator.adapter
+  const feed = await extractFeed(adapter)
+  const value = await fetchData(feed)
+
+  await insertData({ aggregatorId: aggregator.id, feedId: feed.id, value })
+
+  return { value: BigInt(value), aggregator }
+}

core/src/por/main.ts

@@ -0,0 +1,24 @@
+import { logger } from 'ethers'
+import { buildLogger } from '../logger'
+import { POR_AGGREGATOR_HASH } from '../settings'
+import { hookConsoleError } from '../utils'
+import { fetchWithAggregator } from './fetcher'
+import { reportData } from './reporter'
+
+const aggregatorHash = POR_AGGREGATOR_HASH
+const LOGGER = buildLogger()
+
+const main = async () => {
+  hookConsoleError(LOGGER)
+
+  const { value, aggregator } = await fetchWithAggregator(aggregatorHash)
+
+  logger.info(`Fetched data:${value}`)
+
+  await reportData({ value, aggregator, logger: LOGGER })
+}
+
+main().catch((error) => {
+  console.error(error)
+  process.exitCode = 1
+})

core/src/por/reducer.ts

@@ -0,0 +1,71 @@
+import { PorError, PorErrorCode } from './errors'
+
+export const DATA_FEED_REDUCER_MAPPING = {
+  PARSE: parseFn,
+  MUL: mulFn,
+  POW10: pow10Fn,
+  ROUND: roundFn,
+  INDEX: indexFn
+}
+
+/**
+ * Access data in JSON based on given path.
+ *
+ * Example
+ * let obj = {
+ *     RAW: { ETH: { USD: { PRICE: 123 } } },
+ *     DISPLAY: { ETH: { USD: [Object] } }
+ * }
+ * const fn = parseFn(['RAW', 'ETH', 'USD', 'PRICE'])
+ * fn(obj) // return 123
+ */
+export function parseFn(args: string | string[]) {
+  if (typeof args == 'string') {
+    args = args.split(',')
+  }
+
+  function wrapper(obj) {
+    for (const a of args) {
+      if (a in obj) obj = obj[a]
+      else throw new PorError(PorErrorCode.MissingKeyInJson)
+    }
+    return obj
+  }
+  return wrapper
+}
+
+export function mulFn(args: number) {
+  function wrapper(value: number) {
+    return value * args
+  }
+  return wrapper
+}
+
+export function pow10Fn(args: number) {
+  function wrapper(value: number) {
+    return Number(Math.pow(10, args)) * value
+  }
+  return wrapper
+}
+
+export function roundFn() {
+  function wrapper(value: number) {
+    return Math.round(value)
+  }
+  return wrapper
+}
+
+export function indexFn(args: number) {
+  if (args < 0) {
+    throw new PorError(PorErrorCode.IndexOutOfBoundaries)
+  }
+
+  function wrapper(obj) {
+    if (args >= obj.length) {
+      throw new PorError(PorErrorCode.IndexOutOfBoundaries)
+    } else {
+      return obj[args]
+    }
+  }
+  return wrapper
+}

core/src/por/reporter.ts

@@ -0,0 +1,118 @@
+import { Aggregator__factory } from '@bisonai/orakl-contracts'
+import { ethers } from 'ethers'
+import { Logger } from 'pino'
+import { getReporterByOracleAddress } from '../api'
+import { buildWallet, sendTransaction } from '../reporter/utils'
+import {
+  CHAIN,
+  DATA_FEED_FULFILL_GAS_MINIMUM,
+  DATA_FEED_SERVICE_NAME,
+  PROVIDER,
+  PROVIDER_URL
+} from '../settings'
+import { IAggregator, IReporterConfig } from '../types'
+import { buildTransaction } from '../worker/data-feed.utils'
+
+async function shouldReport({
+  aggregator,
+  value,
+  logger
+}: {
+  aggregator: IAggregator
+  value: bigint
+  logger: Logger
+}) {
+  const contract = new ethers.Contract(aggregator.address, Aggregator__factory.abi, PROVIDER)
+  const latestRoundData = await contract.latestRoundData()
+
+  // Check Submission Hearbeat
+  const updatedAt = Number(latestRoundData.updatedAt) * 1000 // convert to milliseconds
+  const timestamp = Date.now()
+  const heartbeat = aggregator.heartbeat
+
+  if (updatedAt + heartbeat < timestamp) {
+    logger.info('Should report by heartbeat check')
+    logger.info(`Last submission time:${updatedAt}, heartbeat:${heartbeat}`)
+    return true
+  }
+
+  // Check deviation threashold
+  if (aggregator.threshold && latestRoundData.answer) {
+    const latestSubmission = Number(latestRoundData.answer)
+    const currentSubmission = Number(value)
+
+    const range = latestSubmission * aggregator.threshold
+    const l = latestSubmission - range
+    const r = latestSubmission + range
+
+    if (currentSubmission < l || currentSubmission > r) {
+      logger.info('Should report by deviation check')
+      logger.info(`Latest submission:${latestSubmission}, currentSubmission:${currentSubmission}`)
+      return true
+    }
+  }
+  return false
+}
+
+async function submit({
+  value,
+  oracleAddress,
+  logger
+}: {
+  value: bigint
+  oracleAddress: string
+  logger: Logger
+}) {
+  const reporter: IReporterConfig = await getReporterByOracleAddress({
+    service: DATA_FEED_SERVICE_NAME,
+    chain: CHAIN,
+    oracleAddress,
+    logger: logger
+  })
+
+  const iface = new ethers.utils.Interface(Aggregator__factory.abi)
+  const contract = new ethers.Contract(oracleAddress, Aggregator__factory.abi, PROVIDER)
+  const queriedRoundId = 0
+  const state = await contract.oracleRoundState(reporter.address, queriedRoundId)
+  const roundId = state._roundId
+
+  const tx = buildTransaction({
+    payloadParameters: {
+      roundId,
+      submission: value
+    },
+    to: oracleAddress,
+    gasMinimum: DATA_FEED_FULFILL_GAS_MINIMUM,
+    iface,
+    logger
+  })
+
+  const wallet = await buildWallet({ privateKey: reporter.privateKey, providerUrl: PROVIDER_URL })
+  const txParams = { wallet, ...tx, logger }
+
+  const NUM_TRANSACTION_TRIALS = 3
+  for (let i = 0; i < NUM_TRANSACTION_TRIALS; ++i) {
+    logger.info(`Reporting to round:${roundId} with value:${value}`)
+    try {
+      await sendTransaction(txParams)
+      break
+    } catch (e) {
+      logger.error('Failed to send transaction')
+      throw e
+    }
+  }
+}
+
+export async function reportData({
+  value,
+  aggregator,
+  logger
+}: {
+  value: bigint
+  aggregator: IAggregator
+  logger: Logger
+}) {
+  if (await shouldReport({ aggregator, value, logger })) {
+    await submit({ value, oracleAddress: aggregator.address, logger })
+  }
+}

core/src/por/types.ts

@@ -0,0 +1,6 @@
+export interface IData {
+  aggregatorId: string
+  feedId: bigint
+  timestamp: string
+  value: number
+}

core/src/settings.ts

@@ -21,6 +21,9 @@ export const SLACK_WEBHOOK_URL = process.env.SLACK_WEBHOOK_URL || ''
 export const LOCAL_AGGREGATOR = process.env.LOCAL_AGGREGATOR || 'MEDIAN'
 export const LISTENER_DELAY = Number(process.env.LISTENER_DELAY) || 500
 
+// POR
+export const POR_AGGREGATOR_HASH = process.env.POR_AGGREGATOR_HASH || ''
+
 // Gas mimimums
 export const VRF_FULFILL_GAS_MINIMUM = 1_000_000
 export const REQUEST_RESPONSE_FULFILL_GAS_MINIMUM = 400_000