Merge pull request #492 from bitfinexcom/staging

Release version to master

Author: batrudinych

bfx-report-ui

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

package-lock.json

@@ -42,6 +42,7 @@ const {
 const appDeps = require('./loc.api/di/app.deps')
 const TYPES = require('./loc.api/di/types')
 const fwDataValidator = require('./loc.api/data-validator')
+const { isBfxApiStaging } = require('./loc.api/helpers')
 
 class WrkReportFrameWorkApi extends WrkReportServiceApi {
   loadCoreDeps (...args) {
@@ -128,8 +129,12 @@ class WrkReportFrameWorkApi extends WrkReportServiceApi {
       syncMode,
       dbDriver,
       verboseSql,
-      dbPathAbsolute
+      dbPathAbsolute,
+      restUrl
     } = this.conf[this.group]
+    const stagingDbLabel = isBfxApiStaging(restUrl)
+      ? '-staging'
+      : ''
     const facs = []
 
     if (syncMode) {
@@ -147,7 +152,7 @@ class WrkReportFrameWorkApi extends WrkReportServiceApi {
           'm0',
           'm0',
           {
-            name: 'sync',
+            name: `sync${stagingDbLabel}`,
             dbPathAbsolute,
             workerPathAbsolute,
             verbose: verboseSql,

package.json

@@ -18,6 +18,7 @@ const {
   getAuthFromSubAccountAuth,
   getSubAccountAuthFromAuth
 } = require('./sub-account-auth')
+const isBfxApiStaging = require('./is-bfx-api-staging')
 
 module.exports = {
   checkParamsAuth,
@@ -33,5 +34,6 @@ module.exports = {
   sumAllObjectsNumbers,
   pickAllLowerObjectsNumbers,
   sumArrayVolumes,
-  pushLargeArr
+  pushLargeArr,
+  isBfxApiStaging
 }

workers/api.framework.report.wrk.js

@@ -0,0 +1,11 @@
+'use strict'
+
+const pattern = /(staging)|(test)/i
+
+module.exports = (restUrl) => {
+  return !!(
+    restUrl &&
+    typeof restUrl === 'string' &&
+    pattern.test(restUrl)
+  )
+}

workers/loc.api/helpers/index.js

@@ -1,7 +1,11 @@
 'use strict'
 
-const { promisify } = require('util')
-const setImmediatePromise = promisify(setImmediate)
+const {
+  setTimeout,
+  setImmediate
+} = require('node:timers/promises')
+const { promisify } = require('node:util')
+
 const MAIN_DB_WORKER_ACTIONS = require(
   'bfx-facs-db-better-sqlite/worker/db-worker-actions/db-worker-actions.const'
 )
@@ -30,7 +34,8 @@ const {
   getGroupQuery,
   getSubQuery,
   getLimitQuery,
-  manageTransaction
+  manageTransaction,
+  isLockedDbError
 } = require('./helpers')
 
 const {
@@ -87,6 +92,10 @@ class BetterSqliteDAO extends DAO {
 
     this._querySet = new Set()
     this._transQuerySet = new Set()
+
+    this._lockedDbRetryTimeout = 300
+    this._lockedDbRetryAmount = 0
+    this._lockedDbRetryInTrxAmount = 0
   }
 
   async restartDb (opts = {}) {
@@ -136,6 +145,19 @@ class BetterSqliteDAO extends DAO {
     } catch (err) {
       this._querySet.delete(newQueryPromise)
 
+      if (
+        isLockedDbError(err) &&
+        this._lockedDbRetryAmount < 3
+      ) {
+        await setTimeout(this._lockedDbRetryTimeout)
+
+        this._lockedDbRetryAmount += 1
+        const res = await this.query(args, opts)
+        this._lockedDbRetryAmount = 0
+
+        return res
+      }
+
       throw err
     }
   }
@@ -170,18 +192,34 @@ class BetterSqliteDAO extends DAO {
 
       return res
     } catch (err) {
-      // Transaction was forcefully rolled back
-      if (!this.db.inTransaction) {
-        throw err
-      }
-      if (isTransBegun) {
+      // if this.db.inTransaction, trx was forcefully rolled back
+      if (
+        this.db.inTransaction &&
+        isTransBegun
+      ) {
         this.db.prepare('ROLLBACK').run()
         isTransBegun = false
       }
-      if (typeof afterTransFn === 'function') {
+      if (
+        this.db.inTransaction &&
+        typeof afterTransFn === 'function'
+      ) {
         await afterTransFn()
       }
 
+      if (
+        isLockedDbError(err) &&
+        this._lockedDbRetryInTrxAmount < 3
+      ) {
+        await setTimeout(this._lockedDbRetryTimeout)
+
+        this._lockedDbRetryInTrxAmount += 1
+        const res = await this._proccesTrans(asyncExecQuery, opts)
+        this._lockedDbRetryInTrxAmount = 0
+
+        return res
+      }
+
       throw err
     }
   }
@@ -559,7 +597,7 @@ class BetterSqliteDAO extends DAO {
         let endId = firstId + chunkLength - 1
 
         while (true) {
-          await setImmediatePromise()
+          await setImmediate()
 
           if (
             !Number.isFinite(startId) ||
@@ -859,7 +897,7 @@ class BetterSqliteDAO extends DAO {
     }
 
     for (const obj of data) {
-      await setImmediatePromise()
+      await setImmediate()
 
       const _obj = mixUserIdToArrData(
         auth,
@@ -894,7 +932,7 @@ class BetterSqliteDAO extends DAO {
 
     await this._beginTrans(async () => {
       for (const { sql, params } of queries) {
-        await setImmediatePromise()
+        await setImmediate()
 
         await this.query({
           action: MAIN_DB_WORKER_ACTIONS.RUN,
@@ -1207,7 +1245,7 @@ class BetterSqliteDAO extends DAO {
     const params = []
 
     for (const obj of data) {
-      await setImmediatePromise()
+      await setImmediate()
 
       const filter = mapObjBySchema(obj, filterPropNames)
       const newItem = mapObjBySchema(obj, upPropNames)
@@ -1232,7 +1270,7 @@ class BetterSqliteDAO extends DAO {
 
     await this._beginTrans(async () => {
       for (const [i, paramsItem] of params.entries()) {
-        await setImmediatePromise()
+        await setImmediate()
 
         await this.query({
           action: MAIN_DB_WORKER_ACTIONS.RUN,
@@ -1252,9 +1290,55 @@ class BetterSqliteDAO extends DAO {
     } = opts ?? {}
     const data = serializeObj(record)
 
-    const res = await this.query({
-      action: DB_WORKER_ACTIONS.UPDATE_RECORD_OF,
-      params: { data, name }
+    const res = await this._beginTrans(async () => {
+      const elems = await this.query({
+        action: MAIN_DB_WORKER_ACTIONS.ALL,
+        sql: `SELECT * FROM ${name}`
+      }, { doNotQueueQuery: true })
+
+      if (
+        !Array.isArray(elems) ||
+        elems.length === 0
+      ) {
+        const keys = Object.keys(data)
+        const projection = getProjectionQuery(keys)
+        const {
+          placeholders,
+          placeholderVal
+        } = getPlaceholdersQuery(data, keys)
+
+        return await this.query({
+          action: MAIN_DB_WORKER_ACTIONS.RUN,
+          sql: `INSERT INTO ${name}(${projection})
+            VALUES (${placeholders})`,
+          params: placeholderVal
+        }, { doNotQueueQuery: true })
+      }
+      if (elems.length > 1) {
+        await this.query({
+          action: MAIN_DB_WORKER_ACTIONS.RUN,
+          sql: `DELETE FROM ${name} WHERE _id != $_id`,
+          params: { _id: elems[0]._id }
+        }, { doNotQueueQuery: true })
+      }
+
+      const { _id } = elems[0] ?? {}
+      const values = { _id }
+      const fields = Object.keys(data)
+        .map((item) => {
+          const key = `new_${item}`
+          values[key] = data[item]
+
+          return `${item} = $${key}`
+        })
+        .join(', ')
+
+      return await this.query({
+        action: MAIN_DB_WORKER_ACTIONS.RUN,
+        sql: `UPDATE ${name} SET ${fields}
+          WHERE _id = $_id`,
+        params: values
+      }, { doNotQueueQuery: true })
     })
 
     if (shouldNotThrowError) {

workers/loc.api/helpers/is-bfx-api-staging.js

@@ -6,6 +6,8 @@ const path = require('path')
 const moment = require('moment')
 const { orderBy } = require('lodash')
 
+const { isBfxApiStaging } = require('../../../helpers')
+
 const { decorateInjectable } = require('../../../di/utils')
 
 const depsTypes = (TYPES) => [
@@ -39,7 +41,10 @@ class DBBackupManager {
       this.conf.dbPathAbsolute,
       'backups'
     )
-    this._dbFileName = 'db-sqlite_sync_m0.db'
+    const stagingDbLabel = isBfxApiStaging(this.conf.restUrl)
+      ? '-staging'
+      : ''
+    this._dbFileName = `db-sqlite_sync${stagingDbLabel}_m0.db`
     this._dbDestination = path.join(
       this.conf.dbPathAbsolute,
       this._dbFileName

workers/loc.api/sync/dao/dao.better.sqlite.js

@@ -36,6 +36,7 @@ const {
   getUsersQuery
 } = require('./users')
 const manageTransaction = require('./manage-transaction')
+const isLockedDbError = require('./is-locked-db-error')
 
 module.exports = {
   mixUserIdToArrData,
@@ -67,5 +68,6 @@ module.exports = {
   fillSubUsers,
   getSubUsersQuery,
   getUsersQuery,
-  manageTransaction
+  manageTransaction,
+  isLockedDbError
 }

workers/loc.api/sync/dao/db.backup.manager/index.js

@@ -0,0 +1,5 @@
+'use strict'
+
+module.exports = (err) => {
+  return err.toString().includes('database is locked')
+}