Merge pull request bitfinexcom#324 from bitfinexcom/staging

Release version to master

package.json

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

test/helpers/helpers.mock-rest-v2.js

@@ -49,7 +49,8 @@ const getMockDataOpts = () => ({
   generate_token: null,
   delete_token: null,
   login: null,
-  login_verify: null
+  login_verify: null,
+  platform_status: null
 })
 
 const getExtraMockMethods = () => (new Map([

test/helpers/mock-data.js

@@ -445,5 +445,9 @@ module.exports = new Map([
         'look at this note'
       ]
     ]
+  ],
+  [
+    'platform_status',
+    [1]
   ]
 ])

test/test-cases/api-sync-mode-sqlite-test-cases.js

@@ -31,6 +31,26 @@ module.exports = (
   } = params
   const auth = { token: '' }
 
+  it('it should be successfully performed by the getPlatformStatus method', async function () {
+    this.timeout(5000)
+
+    const res = await agent
+      .post(`${basePath}/json-rpc`)
+      .type('json')
+      .send({
+        method: 'getPlatformStatus',
+        id: 5
+      })
+      .expect('Content-Type', /json/)
+      .expect(200)
+
+    assert.isObject(res.body)
+    assert.propertyVal(res.body, 'id', 5)
+    assert.isObject(res.body.result)
+    assert.isBoolean(res.body.result.isMaintenance)
+    assert.isNotOk(res.body.result.isMaintenance)
+  })
+
   it('it should be successfully performed by the pingApi method', async function () {
     this.timeout(5000)
 

test/test-cases/get-sync-progress-test-case.js

@@ -32,7 +32,21 @@ module.exports = (
       assert.propertyVal(res.body, 'id', 5)
       assert.isObject(res.body.result)
       assert.isNumber(res.body.result.progress)
+      assert.isBoolean(res.body.result.isSyncInProgress)
 
+      assert.isOk(
+        res.body.result.error === null ||
+        typeof res.body.result.error === 'string'
+      )
+      assert.isOk(
+        res.body.result.state === null ||
+        [
+          'SYNCHRONIZATION_HAS_NOT_BEEN_STARTED_YET',
+          'SYNCHRONIZATION_IS_STARTED',
+          'SYNCHRONIZATION_IS_FINISHED',
+          'SYNCHRONIZATION_HAS_BEEN_INTERRUPTED'
+        ].some((state) => state === res.body.result.state)
+      )
       assert.isOk(
         res.body.result.syncStartedAt === null ||
         Number.isInteger(res.body.result.syncStartedAt)

workers/api.framework.report.wrk.js

@@ -199,7 +199,9 @@ class WrkReportFrameWorkApi extends WrkReportServiceApi {
     await super.stopService()
 
     const wsTransport = this.container.get(TYPES.WSTransport)
+    const processMessageManager = this.container.get(TYPES.ProcessMessageManagerFactory)()
 
+    processMessageManager.stop()
     wsTransport.stop()
   }
 }

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

@@ -54,7 +54,8 @@ class BfxApiRouter extends BaseBfxApiRouter {
       ['fundingCreditHistory', 90],
       ['accountSummary', 90],
       ['logins', 90],
-      ['changeLogs', 90]
+      ['changeLogs', 90],
+      ['status', 30]
     ])
   }
 

workers/loc.api/di/app.deps.js

@@ -160,7 +160,8 @@ module.exports = ({
           ['_dataConsistencyChecker', TYPES.DataConsistencyChecker],
           ['_winLossVSAccountBalance', TYPES.WinLossVSAccountBalance],
           ['_getDataFromApi', TYPES.GetDataFromApi],
-          ['_httpRequest', TYPES.HTTPRequest]
+          ['_httpRequest', TYPES.HTTPRequest],
+          ['_wsEventEmitterFactory', TYPES.WSEventEmitterFactory]
         ]
       })
     rebind(TYPES.RServiceDepsSchemaAliase)

workers/loc.api/generate-csv/csv-writer/full-snapshot-report-csv-writer.js

@@ -1,13 +1,11 @@
 'use strict'
 
-const { pipeline } = require('stream')
-const { stringify } = require('csv')
-
 const {
   write
 } = require('bfx-report/workers/loc.api/queue/write-data-to-stream/helpers')
-
-const nope = () => {}
+const {
+  streamWriter
+} = require('bfx-report/workers/loc.api/generate-csv/csv-writer/helpers')
 
 module.exports = (
   rService,
@@ -35,79 +33,19 @@ module.exports = (
   queue.emit('progress', 0)
 
   if (typeof jobData === 'string') {
-    const stringifier = stringify(
-      { columns: ['mess'] }
+    await streamWriter(
+      wStream,
+      [{
+        columnParams: { columns: ['mess'] },
+        writeFn: (stream) => write([{ mess: jobData }], stream)
+      }]
     )
 
-    pipeline(stringifier, wStream, nope)
-    write([{ mess: jobData }], stringifier)
     queue.emit('progress', 100)
-    stringifier.end()
 
     return
   }
 
-  wStream.setMaxListeners(50)
-
-  const timestampsStringifier = stringify({
-    header: true,
-    columns: columnsCsv.timestamps
-  })
-  const posNameStringifier = stringify(
-    { columns: ['name'] }
-  )
-  const posStringifier = stringify({
-    header: true,
-    columns: columnsCsv.positionsSnapshot
-  })
-  const positionsTotalPlUsdStringifier = stringify({
-    header: true,
-    columns: columnsCsv.positionsTotalPlUsd
-  })
-  const walletsNameStringifier = stringify(
-    { columns: ['name'] }
-  )
-  const walletsStringifier = stringify({
-    header: true,
-    columns: columnsCsv.walletsSnapshot
-  })
-  const walletsTotalBalanceUsdStringifier = stringify({
-    header: true,
-    columns: columnsCsv.walletsTotalBalanceUsd
-  })
-  const positionsTickersNameStringifier = stringify(
-    { columns: ['name'] }
-  )
-  const positionsTickersStringifier = stringify({
-    header: true,
-    columns: columnsCsv.positionsTickers
-  })
-  const walletsTickersNameStringifier = stringify(
-    { columns: ['name'] }
-  )
-  const walletsTickersStringifier = stringify({
-    header: true,
-    columns: columnsCsv.walletsTickers
-  })
-
-  const rStreamArr = [
-    timestampsStringifier,
-    posNameStringifier,
-    posStringifier,
-    positionsTotalPlUsdStringifier,
-    walletsNameStringifier,
-    walletsStringifier,
-    walletsTotalBalanceUsdStringifier,
-    positionsTickersNameStringifier,
-    positionsTickersStringifier,
-    walletsTickersNameStringifier,
-    walletsTickersStringifier
-  ]
-
-  rStreamArr.forEach((rStream) => {
-    pipeline(rStream, wStream, nope)
-  })
-
   const res = await getDataFromApi({
     getData: rService[name].bind(rService),
     args,
@@ -121,65 +59,115 @@ module.exports = (
     walletsTickers,
     positionsTotalPlUsd,
     walletsTotalBalanceUsd
-  } = { ...res }
+  } = res ?? {}
 
-  write(
-    [{ mtsCreated, end }, {}],
-    timestampsStringifier,
-    formatSettings.timestamps,
-    params
-  )
-  write([{ name: 'POSITIONS' }], posNameStringifier)
-  write(
-    [...positionsSnapshot, {}],
-    posStringifier,
-    formatSettings.positionsSnapshot,
-    params
-  )
-  write(
-    [{ plUsd: positionsTotalPlUsd }, {}],
-    positionsTotalPlUsdStringifier,
-    formatSettings.positionsTotalPlUsd,
-    params
-  )
-  write([{ name: 'WALLETS' }], walletsNameStringifier)
-  write(
-    [...walletsSnapshot, {}],
-    walletsStringifier,
-    formatSettings.walletsSnapshot,
-    params
-  )
-  write(
-    [{ balanceUsd: walletsTotalBalanceUsd }, {}],
-    walletsTotalBalanceUsdStringifier,
-    formatSettings.walletsTotalBalanceUsd,
-    params
-  )
-  write([{ name: 'POSITIONS TICKERS' }], positionsTickersNameStringifier)
-  write(
-    [...positionsTickers, {}],
-    positionsTickersStringifier,
-    formatSettings.positionsTickers,
-    params
-  )
-  write([{ name: 'WALLETS TICKERS' }], walletsTickersNameStringifier)
-  write(
-    walletsTickers,
-    walletsTickersStringifier,
-    formatSettings.walletsTickers,
-    params
+  wStream.setMaxListeners(50)
+
+  await streamWriter(
+    wStream,
+    [
+      {
+        columnParams: {
+          header: true,
+          columns: columnsCsv.timestamps
+        },
+        writeFn: (stream) => write(
+          [{ mtsCreated, end }, {}],
+          stream,
+          formatSettings.timestamps,
+          params
+        )
+      },
+      {
+        columnParams: { columns: ['name'] },
+        writeFn: (stream) => write([{ name: 'POSITIONS' }], stream)
+      },
+      {
+        columnParams: {
+          header: true,
+          columns: columnsCsv.positionsSnapshot
+        },
+        writeFn: (stream) => write(
+          [...positionsSnapshot, {}],
+          stream,
+          formatSettings.positionsSnapshot,
+          params
+        )
+      },
+      {
+        columnParams: {
+          header: true,
+          columns: columnsCsv.positionsTotalPlUsd
+        },
+        writeFn: (stream) => write(
+          [{ plUsd: positionsTotalPlUsd }, {}],
+          stream,
+          formatSettings.positionsTotalPlUsd,
+          params
+        )
+      },
+      {
+        columnParams: { columns: ['name'] },
+        writeFn: (stream) => write([{ name: 'WALLETS' }], stream)
+      },
+      {
+        columnParams: {
+          header: true,
+          columns: columnsCsv.walletsSnapshot
+        },
+        writeFn: (stream) => write(
+          [...walletsSnapshot, {}],
+          stream,
+          formatSettings.walletsSnapshot,
+          params
+        )
+      },
+      {
+        columnParams: {
+          header: true,
+          columns: columnsCsv.walletsTotalBalanceUsd
+        },
+        writeFn: (stream) => write(
+          [{ balanceUsd: walletsTotalBalanceUsd }, {}],
+          stream,
+          formatSettings.walletsTotalBalanceUsd,
+          params
+        )
+      },
+      {
+        columnParams: { columns: ['name'] },
+        writeFn: (stream) => write([{ name: 'POSITIONS TICKERS' }], stream)
+      },
+      {
+        columnParams: {
+          header: true,
+          columns: columnsCsv.positionsTickers
+        },
+        writeFn: (stream) => write(
+          [...positionsTickers, {}],
+          stream,
+          formatSettings.positionsTickers,
+          params
+        )
+      },
+      {
+        columnParams: { columns: ['name'] },
+        writeFn: (stream) => write([{ name: 'WALLETS TICKERS' }], stream)
+      },
+      {
+        columnParams: {
+          header: true,
+          columns: columnsCsv.walletsTickers
+        },
+        writeFn: (stream) => write(
+          walletsTickers,
+          stream,
+          formatSettings.walletsTickers,
+          params
+        )
+      }
+    ]
   )
 
   queue.emit('progress', 100)
-
-  posNameStringifier.end()
-  posStringifier.end()
-  positionsTotalPlUsdStringifier.end()
-  walletsNameStringifier.end()
-  walletsStringifier.end()
-  walletsTotalBalanceUsdStringifier.end()
-  positionsTickersNameStringifier.end()
-  positionsTickersStringifier.end()
-  walletsTickersNameStringifier.end()
-  walletsTickersStringifier.end()
 }