Collect exchange rate time series using notification service (#1020)

packages/notification-service/README.md

@@ -20,7 +20,7 @@ Next, run this task to build, configure, and start the service:
 
 Deploy your app. The project will be built automatically by Google Cloud Build:
 
-    yarn deploy:{ENVIRONMENT}
+    yarn deploy -n {ENVIRONMENT}
 
 Current supported environments are production, integration, staging-argentina
 

packages/notification-service/config/config.alfajores.env

@@ -7,4 +7,6 @@ STABLE_TOKEN_ADDRESS=0xd4b4fcaCAc9e23225680e89308E0a4C41Dd9C6B4
 GOLD_TOKEN_ADDRESS=0x11CD75C45638Ec9f41C0e8Df78fc756201E48ff2
 DEFAULT_LOCALE=en
 NOTIFICATION_TTL_MS=604800000
-POLLING_INTERVAL=1000
+POLLING_INTERVAL=1000
+EXCHANGE_POLLING_INTERVAL=1800000
+WEB3_PROVIDER_URL=http://35.185.236.10:8545

packages/notification-service/config/config.integration.env

@@ -8,3 +8,5 @@ GOLD_TOKEN_ADDRESS=0x9102eCD93ac8D66bAc3D397BF52bc57Ee34Bcb87
 DEFAULT_LOCALE=en
 NOTIFICATION_TTL_MS=604800000
 POLLING_INTERVAL=1000
+EXCHANGE_POLLING_INTERVAL=1800000
+WEB3_PROVIDER_URL=http://34.83.137.48:8545

packages/notification-service/config/config.local.env

@@ -9,3 +9,5 @@ DEFAULT_LOCALE=en
 GOOGLE_APPLICATION_CREDENTIALS="./config/serviceAccountKey.json"
 NOTIFICATION_TTL_MS=604800000
 POLLING_INTERVAL=5000
+EXCHANGE_POLLING_INTERVAL=1800000
+WEB3_PROVIDER_URL=http://34.83.137.48:8545

packages/notification-service/package.json

@@ -19,6 +19,8 @@
     "deploy": "./deploy.sh"
   },
   "dependencies": {
+    "@celo/utils": "^0.0.6-beta5",
+    "@celo/walletkit": "^0.0.14",
     "async-polling": "^0.2.1",
     "bignumber.js": "^7.2.0",
     "dotenv": "^6.0.0",
@@ -27,6 +29,7 @@
     "i18next": "^12.1.0",
     "node-fetch": "^2.2.0",
     "utf8": "^3.0.0",
+    "web3": "1.0.0-beta.37",
     "web3-eth-abi": "1.0.0-beta.37",
     "web3-utils": "1.0.0-beta.37"
   },

packages/notification-service/src/config.ts

@@ -36,6 +36,10 @@ export const DEFAULT_LOCALE = process.env.DEFAULT_LOCALE
 export const POLLING_INTERVAL = Number(process.env.POLLING_INTERVAL) || 1000
 export const NOTIFICATIONS_TTL_MS = Number(process.env.NOTIFICATION_TTL_MS) || 3600 * 1000 * 24 * 7 // 1 week in milliseconds
 
+export const EXCHANGE_POLLING_INTERVAL =
+  Number(process.env.EXCHANGE_POLLING_INTERVAL) || 30 * 60 * 1000 // 30 minutes in milliseconds
+export const WEB3_PROVIDER_URL = process.env.WEB3_PROVIDER_URL || 'UNDEFINED'
+
 export enum NotificationTypes {
   PAYMENT_RECEIVED = 'PAYMENT_RECEIVED',
   PAYMENT_REQUESTED = 'PAYMENT_REQUESTED',

packages/notification-service/src/exchange/exchangeQuery.ts

@@ -0,0 +1,45 @@
+import { CURRENCY_ENUM } from '@celo/utils'
+import { ContractUtils } from '@celo/walletkit' // To be updated to contractkit when new version is published on npm
+import BigNumber from 'bignumber.js'
+import Web3 from 'web3'
+import { WEB3_PROVIDER_URL } from '../config'
+import { writeExchangeRatePair } from '../firebase'
+
+// Amounts to estimate the exchange rate, as the rate varies based on transaction size
+const SELL_AMOUNTS = {
+  [CURRENCY_ENUM.DOLLAR]: new BigNumber(10000 * 1000000000000000000), // 100 dollars
+  [CURRENCY_ENUM.GOLD]: new BigNumber(10 * 1000000000000000000), // 10 gold
+}
+
+export async function handleExchangeQuery() {
+  const web3Instance = await getWeb3Instance()
+  const fetchTime = Date.now().toString()
+  const [dollarMakerRate, goldMakerRate] = await Promise.all([
+    getExchangeRate(CURRENCY_ENUM.DOLLAR, web3Instance),
+    getExchangeRate(CURRENCY_ENUM.GOLD, web3Instance),
+  ])
+
+  writeExchangeRatePair(CURRENCY_ENUM.DOLLAR, dollarMakerRate.toString(), fetchTime)
+  writeExchangeRatePair(CURRENCY_ENUM.GOLD, goldMakerRate.toString(), fetchTime)
+}
+
+export async function getExchangeRate(makerToken: CURRENCY_ENUM, web3Instance: Web3) {
+  const rate = await ContractUtils.getExchangeRate(
+    web3Instance,
+    makerToken,
+    SELL_AMOUNTS[makerToken]
+  )
+  return rate
+}
+
+let web3: Web3
+export function getWeb3Instance(): Web3 {
+  if (web3 && web3.eth.net.isListening()) {
+    // Already connected
+    return web3
+  } else {
+    const httpProvider = new Web3.providers.HttpProvider(WEB3_PROVIDER_URL)
+    web3 = new Web3(httpProvider)
+    return web3
+  }
+}

packages/notification-service/src/firebase.ts

@@ -1,3 +1,4 @@
+import { CURRENCY_ENUM } from '@celo/utils'
 import * as admin from 'firebase-admin'
 import i18next from 'i18next'
 import { Currencies } from './blockscout/transfers'
@@ -7,6 +8,7 @@ let database: admin.database.Database
 let registrationsRef: admin.database.Reference
 let lastBlockRef: admin.database.Reference
 let pendingRequestsRef: admin.database.Reference
+let exchangeRatesRef: admin.database.Reference
 
 export interface Registrations {
   [address: string]:
@@ -41,6 +43,12 @@ interface PendingRequests {
   [uid: string]: PaymentRequest
 }
 
+interface ExchangeRateObject {
+  makerToken: CURRENCY_ENUM
+  exchangeRate: string
+  timestamp: string
+}
+
 let registrations: Registrations = {}
 let lastBlockNotified: number = -1
 let pendingRequests: PendingRequests = {}
@@ -68,6 +76,7 @@ export function initializeDb() {
   registrationsRef = database.ref('/registrations')
   lastBlockRef = database.ref('/lastBlockNotified')
   pendingRequestsRef = database.ref('/pendingRequests')
+  exchangeRatesRef = database.ref('/exchangeRates')
 
   // Attach to the registration ref to keep local registrations mapping up to date
   registrationsRef.on(
@@ -139,6 +148,16 @@ export function setPaymentRequestNotified(uid: string): Promise<void> {
   return database.ref(`/pendingRequests/${uid}`).update({ notified: true })
 }
 
+export function writeExchangeRatePair(
+  makerToken: CURRENCY_ENUM,
+  exchangeRate: string,
+  timestamp: string
+) {
+  const exchangeRateRecord: ExchangeRateObject = { makerToken, exchangeRate, timestamp }
+  exchangeRatesRef.push(exchangeRateRecord)
+  console.debug('Recorded exchange rate ', exchangeRateRecord)
+}
+
 export function setLastBlockNotified(newBlock: number): Promise<void> | undefined {
   if (newBlock <= lastBlockNotified) {
     console.debug('Block number less than latest, skipping latestBlock update.')

packages/notification-service/src/index.ts

@@ -1,8 +1,15 @@
 import express from 'express'
 import * as admin from 'firebase-admin'
-import { ENVIRONMENT, FIREBASE_DB, getFirebaseAdminCreds, PORT, VERSION } from './config'
+import {
+  ENVIRONMENT,
+  FIREBASE_DB,
+  getFirebaseAdminCreds,
+  PORT,
+  VERSION,
+  WEB3_PROVIDER_URL,
+} from './config'
 import { getLastBlockNotified, initializeDb as initializeFirebaseDb } from './firebase'
-import { notificationPolling } from './polling'
+import { exchangePolling, notificationPolling } from './polling'
 
 console.info('Service starting with environment, version:', ENVIRONMENT, VERSION)
 const START_TIME = Date.now()
@@ -58,3 +65,13 @@ initializeFirebaseDb()
  */
 console.info('Starting Blockscout polling')
 notificationPolling.run()
+
+if (!WEB3_PROVIDER_URL) {
+  console.info('No Web3 provider found. Skipping exchange polling.')
+} else {
+  /**
+   * Start polling the Exchange contract
+   */
+  console.info('Starting Exchange contract polling')
+  exchangePolling.run()
+}

packages/notification-service/src/polling.ts

@@ -1,6 +1,7 @@
 import AsyncPolling from 'async-polling'
 import { handleTransferNotifications } from './blockscout/transfers'
-import { POLLING_INTERVAL } from './config'
+import { EXCHANGE_POLLING_INTERVAL, POLLING_INTERVAL } from './config'
+import { handleExchangeQuery } from './exchange/exchangeQuery'
 import { handlePaymentRequests } from './handlers'
 
 export const notificationPolling = AsyncPolling(async (end) => {
@@ -13,3 +14,13 @@ export const notificationPolling = AsyncPolling(async (end) => {
     end()
   }
 }, POLLING_INTERVAL)
+
+export const exchangePolling = AsyncPolling(async (end) => {
+  try {
+    await handleExchangeQuery()
+  } catch (e) {
+    console.error('Exchange polling failed', e)
+  } finally {
+    end()
+  }
+}, EXCHANGE_POLLING_INTERVAL)