Merge pull request #279 from bodhiproject/dev-070419

dev-070419

package.json

@@ -3,7 +3,7 @@
   "description": "Server that parses, stores, sends, and receives Bodhi-related data from the blockchain.",
   "author": "bodhi.network",
   "license": "LGPL-3.0",
-  "version": "6.1.3",
+  "version": "6.2.0",
   "repository": "git@github.com:bodhiproject/bodhi-server.git",
   "keywords": [
     "bodhi",
@@ -24,6 +24,7 @@
     "async": "^3.0.1",
     "axios": "^0.18.0",
     "babel-polyfill": "^6.26.0",
+    "bignumber.js": "^9.0.0",
     "chai": "^4.1.2",
     "chai-as-promised": "^7.1.1",
     "crypto-random-string": "^3.0.1",

src/db/db-helper.js

@@ -1,8 +1,12 @@
-const { isNull } = require('lodash');
+const { isNull, each, find } = require('lodash');
 const logger = require('../utils/logger');
 const { isDefined } = require('../utils');
+const { sumBN } = require('../utils/web3-utils');
 const { EVENT_STATUS, TX_STATUS } = require('../constants');
 const { db } = require('.');
+const web3 = require('../web3');
+const EventLeaderboard = require('../models/event-leaderboard');
+const GlobalLeaderboard = require('../models/global-leaderboard');
 
 module.exports = class DBHelper {
   /* Blocks */
@@ -241,16 +245,20 @@ module.exports = class DBHelper {
 
   static async updateEventStatusWithdrawing(currBlockTime) {
     try {
-      await db.Events.update(
+      const updatedEvents = await db.Events.update(
         {
           txStatus: TX_STATUS.SUCCESS,
           $not: { status: EVENT_STATUS.WITHDRAWING },
           arbitrationEndTime: { $lte: currBlockTime },
           currentRound: { $gt: 0 },
         },
         { $set: { status: EVENT_STATUS.WITHDRAWING } },
-        { multi: true },
+        {
+          multi: true,
+          returnUpdatedDocs: true,
+        },
       );
+      return updatedEvents;
     } catch (err) {
       logger.error(`UPDATE Event Status Withdrawing error: ${err.message}`);
       throw err;
@@ -440,4 +448,135 @@ module.exports = class DBHelper {
       throw err;
     }
   }
+
+  /* EventLeaderboard */
+  static async findEventLeaderboard(query, sort) {
+    try {
+      if (sort) return db.EventLeaderboard.cfind(query).sort(sort).exec();
+      return db.EventLeaderboard.find(query);
+    } catch (err) {
+      logger.error(`FIND EventLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  static async findOneEventLeaderboard(query) {
+    try {
+      return db.EventLeaderboard.findOne(query);
+    } catch (err) {
+      logger.error(`FINDONE EventLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  static async countEventLeaderboard(query) {
+    try {
+      return db.EventLeaderboard.count(query);
+    } catch (err) {
+      logger.error(`COUNT EventLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  static async insertEventLeaderboard(entry) {
+    try {
+      const existing = await DBHelper.findOneEventLeaderboard({
+        userAddress: entry.userAddress,
+        eventAddress: entry.eventAddress,
+      });
+      if (isNull(existing)) {
+        await db.EventLeaderboard.insert(entry);
+      } else {
+        const investments = sumBN(existing.investments, entry.investments).toString(10);
+        const winnings = sumBN(existing.winnings, entry.winnings).toString(10);
+        const newEntry = new EventLeaderboard({
+          userAddress: entry.userAddress,
+          eventAddress: entry.eventAddress,
+          investments,
+          winnings,
+        }); // make a new instance, ratio will be calculated in the model
+        await DBHelper.updateEventLeaderboard(newEntry);
+      }
+    } catch (err) {
+      logger.error(`INSERT EventLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  static async updateEventLeaderboard(entry) {
+    try {
+      await db.EventLeaderboard.update(
+        {
+          userAddress: entry.userAddress,
+          eventAddress: entry.eventAddress,
+        },
+        { $set: entry },
+      );
+    } catch (err) {
+      logger.error(`UPDATE EventLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  /* GlobalLeaderboard */
+  static async findGlobalLeaderboard(query, sort) {
+    try {
+      if (sort) return db.GlobalLeaderboard.cfind(query).sort(sort).exec();
+      return db.GlobalLeaderboard.find(query);
+    } catch (err) {
+      logger.error(`FIND GlobalLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  static async findOneGlobalLeaderboard(query) {
+    try {
+      return db.GlobalLeaderboard.findOne(query);
+    } catch (err) {
+      logger.error(`FINDONE GlobalLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  static async countGlobalLeaderboard(query) {
+    try {
+      return db.GlobalLeaderboard.count(query);
+    } catch (err) {
+      logger.error(`COUNT GlobalLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  static async insertGlobalLeaderboard(entry) {
+    try {
+      const existing = await DBHelper.findOneGlobalLeaderboard({ userAddress: entry.userAddress });
+      if (isNull(existing)) {
+        await db.GlobalLeaderboard.insert(entry);
+      } else {
+        const investments = sumBN(existing.investments, entry.investments).toString(10);
+        const winnings = sumBN(existing.winnings, entry.winnings).toString(10);
+        const newEntry = new GlobalLeaderboard({
+          userAddress: entry.userAddress,
+          investments,
+          winnings,
+        }); // make a new instance, ratio will be calculated in the model
+        await DBHelper.updateGlobalLeaderboard(newEntry);
+      }
+    } catch (err) {
+      logger.error(`INSERT GlobalLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
+
+  static async updateGlobalLeaderboard(entry) {
+    try {
+      await db.GlobalLeaderboard.update(
+        { userAddress: entry.userAddress },
+        { $set: entry },
+      );
+    } catch (err) {
+      logger.error(`UPDATE GlobalLeaderboard error: ${err.message}`);
+      throw err;
+    }
+  }
 };

src/db/index.js

@@ -13,6 +13,8 @@ const db = {
   Withdraws: undefined,
   Blocks: undefined,
   TransactionReceipts: undefined,
+  GlobalLeaderboard: undefined,
+  EventLeaderboard: undefined,
 };
 
 /**
@@ -28,6 +30,8 @@ const initDB = async () => {
   db.Withdraws = datastore({ filename: `${dbDir}/withdraws.db` });
   db.Blocks = datastore({ filename: `${dbDir}/blocks.db` });
   db.TransactionReceipts = datastore({ filename: `${dbDir}/transactionreceipts.db` });
+  db.GlobalLeaderboard = datastore({ filename: `${dbDir}/globalleaderboard.db` });
+  db.EventLeaderboard = datastore({ filename: `${dbDir}/eventleaderboard.db` });
 
   try {
     await Promise.all([
@@ -37,6 +41,8 @@ const initDB = async () => {
       db.Withdraws.loadDatabase(),
       db.Blocks.loadDatabase(),
       db.TransactionReceipts.loadDatabase(),
+      db.GlobalLeaderboard.loadDatabase(),
+      db.EventLeaderboard.loadDatabase(),
     ]);
 
     await db.Blocks.ensureIndex({ fieldName: 'blockNum', unique: true });
@@ -45,6 +51,8 @@ const initDB = async () => {
     await db.ResultSets.ensureIndex({ fieldName: 'txid', unique: true });
     await db.Withdraws.ensureIndex({ fieldName: 'txid', unique: true });
     await db.TransactionReceipts.ensureIndex({ fieldName: 'transactionHash' });
+    await db.GlobalLeaderboard.ensureIndex({ fieldName: 'userAddress', unique: true });
+    await db.EventLeaderboard.ensureIndex({ fieldName: 'eventAddress' });
 
     if (process.env.TEST_ENV !== 'true') {
       await applyMigrations();

src/db/migrations/migration2.js

@@ -0,0 +1,107 @@
+const pLimit = require('p-limit');
+const { whilst, eachOfSeries } = require('async');
+const logger = require('../../utils/logger');
+const DBHelper = require('../db-helper');
+const { TX_STATUS, EVENT_STATUS } = require('../../constants');
+const MultipleResultsEventApi = require('../../api/multiple-results-event');
+const EventLeaderboard = require('../../models/event-leaderboard');
+const GlobalLeaderboard = require('../../models/global-leaderboard');
+
+const PROMISE_CONCURRENCY_LIMIT = 15;
+const limit = pLimit(PROMISE_CONCURRENCY_LIMIT);
+
+async function migration2(next) {
+  try {
+    // migrate event leaderboard
+    const events = await DBHelper.findEvent({ txStatus: TX_STATUS.SUCCESS });
+    const promises = [];
+
+    events.forEach((event, key) => {
+      promises.push(limit(async () => {
+        const eventAddress = event.address;
+        const resultSet = await DBHelper.findResultSet({ eventRound: 0, txStatus: TX_STATUS.SUCCESS, eventAddress: event.address });
+        const bets = await DBHelper.findBet({
+          eventAddress: event.address,
+          txStatus: TX_STATUS.SUCCESS,
+        });
+        const txs = resultSet.concat(bets);
+
+        try {
+          let i = 0;
+          await whilst(
+            check => check(null, i < txs.length), // trigger iter
+            async () => {
+              const tx = txs[i];
+              i++;
+              const userAddress = tx.betterAddress || tx.centralizedOracleAddress;
+              const eventLeaderboardEntry = new EventLeaderboard({
+                eventAddress,
+                userAddress,
+                investments: tx.amount,
+                winnings: '0',
+              });
+              await DBHelper.insertEventLeaderboard(eventLeaderboardEntry);
+              if (event.status === EVENT_STATUS.WITHDRAWING) {
+                const globalLeaderboardEntry = new GlobalLeaderboard({
+                  userAddress,
+                  investments: tx.amount,
+                  winnings: '0',
+                });
+                await DBHelper.insertGlobalLeaderboard(globalLeaderboardEntry);
+              }
+            },
+          );
+        } catch (err) {
+        // will only be called if should end this loop
+          logger.error(err);
+          throw err;
+        }
+
+        if (event.status === EVENT_STATUS.WITHDRAWING) {
+          const addresses = new Set();
+          txs.forEach((tx, index) => {
+            if (tx.resultIndex !== event.currentResultIndex) return;
+            if (tx.betterAddress) addresses.add(tx.betterAddress);
+            if (tx.centralizedOracleAddress) addresses.add(tx.centralizedOracleAddress);
+          });
+          await eachOfSeries(addresses, async (address, index) => {
+            try {
+              // calculate winning for this user in this event
+              const res = await MultipleResultsEventApi.calculateWinnings({
+                eventAddress,
+                address,
+              });
+              const winnings = res.toString(10);
+              // update event leaderboard winnings
+              const eventLeaderboardEntry = new EventLeaderboard({
+                eventAddress,
+                userAddress: address,
+                investments: '0',
+                winnings,
+              });
+              await DBHelper.insertEventLeaderboard(eventLeaderboardEntry);
+              const globalLeaderboardEntry = new GlobalLeaderboard({
+                userAddress: address,
+                investments: '0',
+                winnings,
+              });
+              await DBHelper.insertGlobalLeaderboard(globalLeaderboardEntry);
+            } catch (err) {
+              logger.error('Migrate event leaderboard error:', err);
+              throw err;
+            }
+          });
+        }
+      }));
+    });
+
+    await Promise.all(promises);
+    logger.info('Migration 2 done');
+    next();
+  } catch (err) {
+    logger.error(`Migration 2 error: ${err.message}`);
+    throw err;
+  }
+}
+
+module.exports = migration2;

src/graphql/queries/index.js

@@ -10,6 +10,7 @@ const allStats = require('./all-stats');
 const mostBets = require('./most-bets');
 const biggestWinners = require('./biggest-winners');
 const withdrawableEvents = require('./withdrawable-events');
+const { eventLeaderboardEntries, globalLeaderboardEntries } = require('./leaderboard');
 
 module.exports = {
   events,
@@ -24,4 +25,6 @@ module.exports = {
   mostBets,
   biggestWinners,
   withdrawableEvents,
+  eventLeaderboardEntries,
+  globalLeaderboardEntries,
 };