Fabo/rewards per era (#563)

* initial working state * working, this time is true ;-) * fix validator, comments * fix, cleanup * fix validator * update comment * Use coinReducer * cleanup * cleanup unused vars * cleanup * Flatten era rewards and then aggregate per valid * refactored * typo * uüdated polkadot * load rewards * steps towards storing the rewards * fix era format * fix validator in row needs to be address * fix gql query * format bug * some refactoring * fix missing await * correctly get api * push validator getter inside sources * added address to lunie reward internaly * add address correctly * fix variable naming * filter empty rewards * add height from rewards * fixed tests * also fixed terra getRewards here * Update lib/block-listeners/polkadot-node-subscription.js Co-Authored-By: Fabian <[email protected]> * remove dead code * fix stuff * fix operatorAddress of undefined * readded vanished code * don't store validatorAddress Co-authored-by: mariopino <[email protected]> Co-authored-by: Bitcoinera <[email protected]> Co-authored-by: Ana G <[email protected]>
luniehq · Apr 6, 2020 · 45512bf · 45512bf
1 parent a9e1829
commit 45512bf
Show file tree

Hide file tree

Showing 8 changed files with 122 additions and 36 deletions.
diff --git a/lib/block-listeners/polkadot-node-subscription.js b/lib/block-listeners/polkadot-node-subscription.js
@@ -25,7 +25,9 @@ class PolkadotNodeSubscription {
     this.db = new database(config)(networkSchemaName)
     this.height = 0
     this.currentSessionIndex = 0
+    this.currentEra = 0
     this.blockQueue = []
+    this.chainId = `kusama-cc3` // TODO make dynamic
     this.subscribeForNewBlock()
   }
 
@@ -123,7 +125,26 @@ class PolkadotNodeSubscription {
           `\x1b[36mCurrent session index is ${block.sessionIndex}, fetching validators!\x1b[0m`
         )
         this.currentSessionIndex = block.sessionIndex
-        this.sessionValidators = await this.polkadotAPI.getAllValidators()
+        const [sessionValidators, era] = await Promise.all([
+          this.polkadotAPI.getAllValidators(),
+          this.api.query.staking.currentEra().then(async era => {
+            return Number(era.toHuman())
+          })
+        ])
+        this.sessionValidators = sessionValidators
+
+        if (this.currentEra < era || this.currentEra === 0) {
+          console.log(
+            `\x1b[36mCurrent staking era is ${era}, fetching rewards!\x1b[0m`
+          )
+          this.currentEra = era
+
+          if (!(await this.eraRewardsExist(era))) {
+            this.updateRewards(era, this.sessionValidators[0].chainId).catch(
+              console.error
+            )
+          }
+        }
       }
 
       this.updateDBValidatorProfiles(this.sessionValidators)
@@ -146,7 +167,7 @@ class PolkadotNodeSubscription {
         })
       })
     } catch (error) {
-      console.error(`newBlockHandler failed`, JSON.stringify(error, null, 2))
+      console.error(`newBlockHandler failed`, error.message)
       Sentry.captureException(error)
     }
   }
@@ -180,7 +201,7 @@ class PolkadotNodeSubscription {
         name: v.name
       }))
     ]
-    // update only new onces
+    // update only new ones
     const validatorRows = newValidators.map(
       ({ operatorAddress, name, chainId }) => ({
         operator_address: operatorAddress,
@@ -190,6 +211,43 @@ class PolkadotNodeSubscription {
     )
     return this.db.upsert('validatorprofiles', validatorRows)
   }
+  async updateRewards(era, chainId) {
+    console.time()
+    console.log('update rewards')
+    const rewards = await this.polkadotAPI.getEraRewards(era) // ATTENTION: era means "get all rewards of all eras since that era". putting a low number takes a long time.
+    console.timeEnd()
+
+    console.time()
+    console.log('store rewards', rewards.length)
+    await this.storeRewards(
+      rewards
+        .filter(({ amount }) => amount > 0)
+        .map(({ amount, height, validatorAddress, denom, address }) => ({
+          amount,
+          height,
+          validator: validatorAddress,
+          denom,
+          address
+        })),
+      era,
+      chainId
+    )
+    console.timeEnd()
+  }
+
+  storeRewards(rewards, chainId) {
+    return this.db.insert('rewards', rewards, undefined, chainId) // height is in the rewards rows already
+  }
+
+  async eraRewardsExist(era) {
+    const response = await this.db.read(
+      'rewards',
+      'rewardsExitCheck',
+      ['amount'],
+      `limit:1 where:{height:{_eq:"${era}"}}`
+    )
+    return response.length > 0
+  }
 }
 
 module.exports = PolkadotNodeSubscription
diff --git a/lib/reducers/polkadotV0-reducers.js b/lib/reducers/polkadotV0-reducers.js
@@ -194,11 +194,11 @@ function extractInvolvedAddresses(extrinsic) {
 }
 
 // Flatten era rewards and then aggregate per validator
-function rewardsReducer(network, validatorsDictionary, rewards, reducers) {
+function rewardsReducer(network, validators, rewards, reducers) {
   return rewards.reduce((collection, reward) => {
     const validatorRewards = reducers.rewardReducer(
       network,
-      validatorsDictionary,
+      validators,
       reward,
       reducers
     )
@@ -219,12 +219,13 @@ function rewardsReducer(network, validatorsDictionary, rewards, reducers) {
   }, [])
 }
 
-function rewardReducer(network, validatorsDictionary, reward, reducers) {
+function rewardReducer(network, validators, reward, reducers) {
   let parsedRewards = []
   Object.entries(reward.validators).map(validatorReward => {
     const reward = {
       ...reducers.coinReducer(network, validatorReward[1].toString()),
-      validator: validatorsDictionary[validatorReward[0].toString()]
+      validator: validators[validatorReward[0]],
+      validatorAddress: validatorReward[0] // added for writing the validator to the db even it it is not in the dictionary
     }
     parsedRewards.push(reward)
   })

diff --git a/lib/resolvers.js b/lib/resolvers.js
@@ -189,10 +189,8 @@ const resolvers = {
       { dataSources }
     ) => {
       await localStore(dataSources, networkId).dataReady
-      const validatorsDictionary = localStore(dataSources, networkId).validators
       return remoteFetch(dataSources, networkId).getRewards(
         address,
-        validatorsDictionary,
         fiatCurrency
       )
     },
@@ -202,10 +200,8 @@ const resolvers = {
       { dataSources }
     ) => {
       await localStore(dataSources, networkId).dataReady
-      const validatorsDictionary = localStore(dataSources, networkId).validators
       const rewards = await remoteFetch(dataSources, networkId).getRewards(
         address,
-        validatorsDictionary,
         fiatCurrency
       )
       const stakingDenom = await remoteFetch(
@@ -349,10 +345,8 @@ const resolvers = {
       { dataSources }
     ) => {
       await localStore(dataSources, networkId).dataReady
-      const validatorsDictionary = localStore(dataSources, networkId).validators
       let rewards = await remoteFetch(dataSources, networkId).getRewards(
         delegatorAddress,
-        validatorsDictionary,
         fiatCurrency
       )
 

diff --git a/lib/source/cosmosV0-source.js b/lib/source/cosmosV0-source.js
@@ -453,11 +453,11 @@ class CosmosV0API extends RESTDataSource {
     return expectedReturns
   }
 
-  async getRewards(delegatorAddress, validatorsDictionary) {
+  async getRewards(delegatorAddress) {
     this.checkAddress(delegatorAddress)
     const delegations = await this.getDelegationsForDelegatorAddress(
       delegatorAddress,
-      validatorsDictionary
+      this.store.validators
     )
     const rewards = await Promise.all(
       delegations.map(async ({ validatorAddress, validator }) => ({

diff --git a/lib/source/cosmosV2-source.js b/lib/source/cosmosV2-source.js
@@ -150,7 +150,7 @@ class CosmosV2API extends CosmosV0API {
       : []
   }
 
-  async getRewards(delegatorAddress, validatorsDictionary) {
+  async getRewards(delegatorAddress) {
     this.checkAddress(delegatorAddress)
     const result = await this.query(
       `distribution/delegators/${delegatorAddress}/rewards`
@@ -161,7 +161,7 @@ class CosmosV2API extends CosmosV0API {
       .map(({ reward, validator_address }) =>
         this.reducers.rewardReducer(
           reward[0],
-          validatorsDictionary[validator_address]
+          this.store.validators[validator_address]
         )
       )
   }

diff --git a/lib/source/polkadotV0-source.js b/lib/source/polkadotV0-source.js
@@ -159,13 +159,14 @@ class polkadotAPI {
     return null
   }
 
-  async getRewards(delegatorAddress, validatorsDictionary) {
+  async getRewards(delegatorAddress) {
     // This is currently super slow if the account have many pending payouts (same as in Polkadot JS UI)
     // Example execution times:
     // - 10-15s for an account with 1 pending payout
     // - Hangs API for an account with >= 20 eras pending payouts
     console.time(`getRewards`)
     const api = await this.getAPI()
+    const validators = this.store.validators
     const stakingInfo = await api.derive.staking.query(delegatorAddress)
     const lastEraReward = parseInt(stakingInfo.stakingLedger.lastReward) + 1
     const rewards = await api.derive.staking.stakerRewards(
@@ -174,7 +175,45 @@ class polkadotAPI {
     )
     console.timeEnd(`getRewards`)
     return this.reducers.rewardsReducer(
-      validatorsDictionary,
+      this.network,
+      validators,
+      rewards,
+      this.reducers
+    )
+  }
+
+  getAllDelegators() {
+    const validators = this.store.validators
+    const delegators = []
+    Object.values(validators).forEach(validator => {
+      validator.nominations.forEach(nomination => {
+        if (!delegators.find(address => address === nomination.who)) {
+          delegators.push(nomination.who)
+        }
+      })
+    })
+
+    return delegators
+  }
+
+  async getEraRewards(era) {
+    const api = await this.getAPI()
+    const delegators = this.getAllDelegators()
+
+    const nestedRewards = await Promise.all(
+      delegators.map(async address => {
+        const rewards = await api.derive.staking.stakerRewards(address, era - 1)
+        return rewards.map(reward => ({
+          ...reward,
+          address
+        }))
+      })
+    )
+    const rewards = [].concat(...nestedRewards)
+
+    return this.reducers.rewardsReducer(
+      this.network,
+      this.store.validators,
       rewards,
       this.reducers
     )

diff --git a/lib/source/terraV3-source.js b/lib/source/terraV3-source.js
@@ -123,7 +123,7 @@ class TerraV3API extends CosmosV2API {
   }
 
   // Terra will be the root source for functions specific to Tendermint multidenom networks
-  async getRewards(delegatorAddress, validatorsDictionary, fiatCurrency) {
+  async getRewards(delegatorAddress, fiatCurrency) {
     this.checkAddress(delegatorAddress)
     const result = await this.query(
       `distribution/delegators/${delegatorAddress}/rewards`
@@ -136,7 +136,7 @@ class TerraV3API extends CosmosV2API {
       : rewards
     return this.reducers.rewardReducer(
       rewardsWithFiatValue,
-      validatorsDictionary
+      this.store.validators
     )
   }
 

diff --git a/tests/source/cosmosv2-source.test.js b/tests/source/cosmosv2-source.test.js
@@ -28,22 +28,22 @@ jest.mock('apollo-datasource-rest', () => {
 
 describe('Cosmos V2 API', function() {
   describe('getRewards()', function() {
-    let api, cosmosNetworkConfig
+    let api, cosmosNetworkConfig, store
 
     beforeEach(() => {
       cosmosNetworkConfig = networks.find(
         network => network.id === 'cosmos-hub-testnet'
       )
-      api = new CosmosV2API(cosmosNetworkConfig, {})
+      store = {
+        validators: mockValidatorsDictionary
+      }
+      api = new CosmosV2API(cosmosNetworkConfig, store)
       mockDelegatorRewards = JSON.parse(JSON.stringify(delegatorRewards))
     })
 
     it('When an existing delegator address is passed, it should return the calculated rewards', async () => {
       //Act
-      const result = await api.getRewards(
-        delegatorAddress,
-        mockValidatorsDictionary
-      )
+      const result = await api.getRewards(delegatorAddress)
 
       //Assert
       expect(result[0]).toHaveProperty('amount')
@@ -58,19 +58,15 @@ describe('Cosmos V2 API', function() {
       mockDelegatorRewards.result.rewards = []
 
       //Act & Assert
-      await expect(
-        api.getRewards(delegatorAddress, mockValidatorsDictionary)
-      ).resolves.toEqual([])
+      await expect(api.getRewards(delegatorAddress)).resolves.toEqual([])
     })
 
     it('When an existing delegator address is passed with a reward 49000000 (umuon), it should return amount 49 (muon)', async () => {
       //Arrange
       mockDelegatorRewards.result.rewards[0].reward[0].amount = 49000000
 
       //Act & Assert
-      await expect(
-        api.getRewards(delegatorAddress, mockValidatorsDictionary)
-      ).resolves.toEqual([
+      await expect(api.getRewards(delegatorAddress)).resolves.toEqual([
         {
           amount: '49.000000',
           denom: 'MUON',
@@ -84,9 +80,7 @@ describe('Cosmos V2 API', function() {
       mockDelegatorRewards.result.rewards[0].reward[0].amount = 0.05
 
       //Act & Assert
-      await expect(
-        api.getRewards(delegatorAddress, mockValidatorsDictionary)
-      ).resolves.toEqual([
+      await expect(api.getRewards(delegatorAddress)).resolves.toEqual([
         {
           amount: '0.000000',
           denom: 'MUON',