Merge pull request #339 from bitfinexcom/staging

Release Master

package.json

@@ -1,6 +1,6 @@
 {
   "name": "bfx-reports-framework",
-  "version": "4.11.0",
+  "version": "4.12.0",
   "description": "Bitfinex reports framework",
   "main": "worker.js",
   "license": "Apache-2.0",

workers/loc.api/bfx.api.router/index.js

@@ -41,9 +41,9 @@ class BfxApiRouter extends BaseBfxApiRouter {
       ['payInvoiceList', 90],
       ['accountTrades', 90],
       ['fundingTrades', 90],
-      ['trades', 15],
+      ['trades', 10],
       ['statusMessages', 90],
-      ['candles', 30],
+      ['candles', 20],
       ['orderTrades', 90],
       ['orderHistory', 90],
       ['activeOrders', 90],

workers/loc.api/di/factories/sync-user-step-data-factory.js

@@ -9,6 +9,7 @@ module.exports = (ctx) => {
     )
 
     const paramsOrder = [
+      'syncedAt',
       'symbol',
       'timeframe',
       'baseStart',

workers/loc.api/sync/authenticator/index.js

@@ -1082,19 +1082,25 @@ class Authenticator {
     } = user ?? {}
     const tokenKey = this._getTokenKeyByEmailField(user)
 
+    const userSession = this.userSessions.get(token) ?? {}
     const authTokenRefreshInterval = authToken
       ? this.setupAuthTokenRefreshInterval(user)
       : null
 
     this.userSessions.set(
-      token, {
-        ...user,
-        authTokenFn: () => {
-          return this.userSessions.get(token)?.authToken
-        },
-        authTokenRefreshInterval,
-        authTokenInvalidateIntervals: new Map()
-      }
+      token,
+      Object.assign(
+        userSession,
+        user,
+        {
+          authTokenFn: () => {
+            return this.userSessions.get(token)?.authToken
+          },
+          authTokenRefreshInterval,
+          authTokenInvalidateIntervals: userSession
+            ?.authTokenInvalidateIntervals ?? new Map()
+        }
+      )
     )
     this.userTokenMapByEmail.set(tokenKey, token)
   }
@@ -1133,9 +1139,8 @@ class Authenticator {
       ? token
       : this.userTokenMapByEmail.get(tokenKey)
 
-    this.userTokenMapByEmail.delete(tokenKey)
-
     const session = this.userSessions.get(_token) ?? {}
+    this.userTokenMapByEmail.delete(this._getTokenKeyByEmailField(session))
     const {
       authTokenRefreshInterval,
       authTokenInvalidateIntervals = new Map()
@@ -1272,8 +1277,12 @@ class Authenticator {
     const { authTokenInvalidateIntervals } = userSession
     let count = 0
 
+    if (authTokenInvalidateIntervals.has(authToken)) {
+      return
+    }
+
     const authTokenInvalidateInterval = setInterval(async () => {
-      const session = this.userSessions.get(token)
+      const session = this.userSessions.get(token) ?? userSession
 
       try {
         count += 1
@@ -1288,7 +1297,7 @@ class Authenticator {
       } catch (err) {
         if (count >= 3) {
           clearInterval(authTokenInvalidateInterval)
-          session.authTokenInvalidateIntervals?.delete(authToken)
+          session?.authTokenInvalidateIntervals?.delete(authToken)
         }
 
         this.logger.debug(err)

workers/loc.api/sync/currency.converter/index.js

@@ -30,7 +30,8 @@ const depsTypes = (TYPES) => [
   TYPES.SyncSchema,
   TYPES.FOREX_SYMBS,
   TYPES.ALLOWED_COLLS,
-  TYPES.SYNC_API_METHODS
+  TYPES.SYNC_API_METHODS,
+  TYPES.Logger
 ]
 class CurrencyConverter {
   constructor (
@@ -40,7 +41,8 @@ class CurrencyConverter {
     syncSchema,
     FOREX_SYMBS,
     ALLOWED_COLLS,
-    SYNC_API_METHODS
+    SYNC_API_METHODS,
+    logger
   ) {
     this.rService = rService
     this.getDataFromApi = getDataFromApi
@@ -49,6 +51,7 @@ class CurrencyConverter {
     this.FOREX_SYMBS = FOREX_SYMBS
     this.ALLOWED_COLLS = ALLOWED_COLLS
     this.SYNC_API_METHODS = SYNC_API_METHODS
+    this.logger = logger
 
     this._COLL_NAMES = {
       PUBLIC_TRADES: 'publicTrades',
@@ -114,6 +117,8 @@ class CurrencyConverter {
           return this.currenciesSynonymous
         }
       } catch (err) {
+        this.logger.debug(err)
+
         return this.currenciesSynonymous
       }
     }

workers/loc.api/sync/data.inserter/data.checker/index.js

@@ -95,14 +95,17 @@ class DataChecker {
     return filterMethodCollMap(methodCollMap)
   }
 
-  async checkNewPublicData () {
+  /*
+   * `authMap` can be empty
+   */
+  async checkNewPublicData (authMap) {
     const methodCollMap = this._getMethodCollMap()
 
     if (this._isInterrupted) {
       return filterMethodCollMap(methodCollMap, true)
     }
 
-    await this._checkNewDataPublicArrObjType(methodCollMap)
+    await this._checkNewDataPublicArrObjType(authMap, methodCollMap)
     await this._checkNewPublicUpdatableData(methodCollMap)
 
     return filterMethodCollMap(methodCollMap, true)
@@ -179,7 +182,7 @@ class DataChecker {
     schema.start.push(freshSyncUserStepData)
   }
 
-  async _checkNewDataPublicArrObjType (methodCollMap) {
+  async _checkNewDataPublicArrObjType (authMap, methodCollMap) {
     for (const [method, schema] of methodCollMap) {
       if (this._isInterrupted) {
         return
@@ -191,7 +194,21 @@ class DataChecker {
       this._resetSyncSchemaProps(schema)
 
       if (schema.name === this.ALLOWED_COLLS.CANDLES) {
-        await this._checkNewCandlesData(method, schema)
+        // If `authMap` is empty sync candles for all users
+        const _authMap = (
+          !(authMap instanceof Map) ||
+          authMap.size === 0
+        )
+          ? new Map([['ALL', {}]])
+          : authMap
+
+        for (const authItem of _authMap) {
+          await this._checkNewCandlesData(
+            method,
+            schema,
+            authItem[1]
+          )
+        }
       }
       if (
         schema.name === this.ALLOWED_COLLS.PUBLIC_TRADES ||
@@ -336,20 +353,29 @@ class DataChecker {
    */
   async _checkNewCandlesData (
     method,
-    schema
+    schema,
+    auth
   ) {
     if (this._isInterrupted) {
       return
     }
 
+    const { _id: userId, subUser } = auth ?? {}
+    const { _id: subUserId } = subUser ?? {}
+    const usersFilter = Number.isInteger(userId)
+      ? { $eq: { user_id: userId } }
+      : {}
+
     const currMts = Date.now()
-    const firstLedgerMts = await this._getFirstLedgerMts()
+    const firstLedgerMts = await this._getFirstLedgerMts(usersFilter)
 
     if (!Number.isInteger(firstLedgerMts)) {
       return
     }
 
-    const uniqueSymbsSet = await this._getUniqueSymbsFromLedgers()
+    const uniqueSymbsSet = await this._getUniqueSymbsFromLedgers(
+      usersFilter
+    )
     const candlesPairsSet = new Set()
 
     for (const symbol of uniqueSymbsSet) {
@@ -389,7 +415,9 @@ class DataChecker {
           collName: method,
           symbol,
           timeframe: CANDLES_TIMEFRAME,
-          defaultStart: firstLedgerMts
+          defaultStart: firstLedgerMts,
+          userId,
+          subUserId
         }
       )
 
@@ -398,7 +426,6 @@ class DataChecker {
         !syncUserStepData.isCurrStepReady
       ) {
         schema.hasNewData = true
-        schema.start.push(syncUserStepData)
       }
 
       const wasStartPointChanged = this._wasStartPointChanged(
@@ -407,39 +434,42 @@ class DataChecker {
       )
       const shouldFreshSyncBeAdded = this._shouldFreshSyncBeAdded(
         syncUserStepData,
-        currMts
+        currMts,
+        { dayOfYear: 1 }
       )
 
-      if (
-        !wasStartPointChanged &&
-        !shouldFreshSyncBeAdded
-      ) {
-        continue
-      }
-
-      const freshSyncUserStepData = this.syncUserStepDataFactory({
-        ...syncUserStepData.getParams(),
-        isBaseStepReady: true,
-        isCurrStepReady: true
-      })
-
       if (wasStartPointChanged) {
-        freshSyncUserStepData.setParams({
+        syncUserStepData.setParams({
           baseStart: firstLedgerMts,
-          baseEnd: syncUserStepData.baseStart,
           isBaseStepReady: false
         })
+
+        schema.hasNewData = true
       }
       if (shouldFreshSyncBeAdded) {
-        freshSyncUserStepData.setParams({
+        syncUserStepData.setParams({
           currStart: lastElemMtsFromTables,
           currEnd: currMts,
           isCurrStepReady: false
         })
+
+        schema.hasNewData = true
       }
 
-      schema.hasNewData = true
-      schema.start.push(freshSyncUserStepData)
+      if (!schema.hasNewData) {
+        continue
+      }
+
+      // To keep flow: one candles request per currency
+      if (!syncUserStepData.isBaseStepReady) {
+        syncUserStepData.setParams({
+          baseEnd: syncUserStepData.currEnd ?? currMts,
+          isCurrStepReady: true
+        })
+      }
+
+      syncUserStepData.auth = auth
+      schema.start.push(syncUserStepData)
     }
   }
 
@@ -502,7 +532,8 @@ class DataChecker {
   ) {
     const {
       measure = 'minutes',
-      allowedTimeDiff = 60
+      allowedTimeDiff = 60,
+      dayOfYear
     } = allowedDiff ?? {}
 
     const baseEnd = (
@@ -517,15 +548,28 @@ class DataChecker {
     )
       ? syncUserStepData.currEnd
       : 0
+    const syncedAt = syncUserStepData.hasSyncedAt
+      ? syncUserStepData.syncedAt
+      : 0
 
     const momentBaseEnd = moment.utc(baseEnd)
     const momentCurrEnd = moment.utc(currEnd)
     const momentCurrMts = moment.utc(currMts)
+    const momentSyncedAt = moment.utc(syncedAt)
 
     const momentMaxEnd = moment.max(momentBaseEnd, momentCurrEnd)
     const momentDiff = momentCurrMts.diff(momentMaxEnd, measure)
-
-    return momentDiff > allowedTimeDiff
+    const yearsDiff = momentCurrMts.year() - momentSyncedAt.year()
+    const dayOfYearDiff = momentCurrMts.dayOfYear() - momentSyncedAt.dayOfYear()
+
+    return (
+      momentDiff > allowedTimeDiff &&
+      (
+        !Number.isFinite(dayOfYear) ||
+        yearsDiff >= 1 ||
+        dayOfYearDiff >= dayOfYear
+      )
+    )
   }
 
   _wasStartPointChanged (
@@ -553,8 +597,11 @@ class DataChecker {
     return momentDiff > allowedTimeDiff
   }
 
-  async _getFirstLedgerMts () {
-    const firstElemFilter = { $not: { currency: 'USD' } }
+  async _getFirstLedgerMts (usersFilter) {
+    const firstElemFilter = {
+      ...usersFilter,
+      $not: { currency: 'USD' }
+    }
     const firstElemOrder = [['mts', 1]]
 
     const tempLedgersTableName = SyncTempTablesManager.getTempTableName(
@@ -600,9 +647,12 @@ class DataChecker {
     return mts
   }
 
-  async _getUniqueSymbsFromLedgers () {
+  async _getUniqueSymbsFromLedgers (usersFilter) {
     const ledgerParams = {
-      filter: { $not: { currency: this.FOREX_SYMBS } },
+      filter: {
+        ...usersFilter,
+        $not: { currency: this.FOREX_SYMBS }
+      },
       isDistinct: true,
       projection: ['currency']
     }