Release version to master

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@bitfinex/bfx-report",
-  "version": "5.2.1",
+  "version": "5.2.2",
   "description": "Reporting tool",
   "main": "worker.js",
   "engines": {

test/helpers/helpers.core.js

@@ -1,13 +1,16 @@
 'use strict'
 
-const path = require('path')
-
+const path = require('node:path')
 const {
   readdir,
   mkdir,
   rm
 } = require('node:fs/promises')
 
+const QUEUE_EVENT_NAMES = require(
+  '../../workers/loc.api/queue/queue.event.names'
+)
+
 const rmDB = async (
   dir,
   exclude = ['.gitkeep']
@@ -55,9 +58,9 @@ const rmAllFiles = async (dir, exclude) => {
 
 const queueToPromise = (queue) => {
   return new Promise((resolve, reject) => {
-    queue.once('error:base', reject)
-    queue.once('completed', res => {
-      queue.removeListener('error:base', reject)
+    queue.once(QUEUE_EVENT_NAMES.ERROR_BASE, reject)
+    queue.once(QUEUE_EVENT_NAMES.COMPLETED, res => {
+      queue.removeListener(QUEUE_EVENT_NAMES.ERROR_BASE, reject)
       resolve(res)
     })
   })
@@ -77,14 +80,14 @@ const queueToPromiseMulti = (queue, count, cb = () => { }) => {
       }
 
       if (currCount >= count) {
-        queue.removeListener('completed', onCompleted)
-        queue.removeListener('error:base', reject)
+        queue.removeListener(QUEUE_EVENT_NAMES.COMPLETED, onCompleted)
+        queue.removeListener(QUEUE_EVENT_NAMES.ERROR_BASE, reject)
         resolve()
       }
     }
 
-    queue.once('error:base', reject)
-    queue.on('completed', onCompleted)
+    queue.once(QUEUE_EVENT_NAMES.ERROR_BASE, reject)
+    queue.on(QUEUE_EVENT_NAMES.COMPLETED, onCompleted)
   })
 }
 
@@ -103,17 +106,17 @@ const queuesToPromiseMulti = (queues, count, cb = () => { }) => {
 
       if (currCount >= count) {
         queues.forEach(queue => {
-          queue.removeListener('completed', onCompleted)
-          queue.removeListener('error:base', reject)
+          queue.removeListener(QUEUE_EVENT_NAMES.COMPLETED, onCompleted)
+          queue.removeListener(QUEUE_EVENT_NAMES.ERROR_BASE, reject)
         })
 
         resolve()
       }
     }
 
     queues.forEach(queue => {
-      queue.once('error:base', reject)
-      queue.on('completed', onCompleted)
+      queue.once(QUEUE_EVENT_NAMES.ERROR_BASE, reject)
+      queue.on(QUEUE_EVENT_NAMES.COMPLETED, onCompleted)
     })
   })
 }

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

@@ -241,7 +241,10 @@ const createMockRESTv2SrvWithDate = (
         typeof mockData[0] !== 'object'
       )
     ) {
-      srv.setResponse(key, [...mockData])
+      const _mockData = Array.isArray(mockData)
+        ? [...mockData]
+        : mockData
+      srv.setResponse(key, _mockData)
 
       return
     }

test/helpers/helpers.queue-load.js

@@ -6,6 +6,10 @@ require('events')
 
 const argv = require('yargs').argv
 
+const QUEUE_EVENT_NAMES = require(
+  '../../workers/loc.api/queue/queue.event.names'
+)
+
 const {
   startEnvironment,
   stopEnvironment
@@ -49,18 +53,18 @@ process.on('SIGTERM', _stop)
     const processorQueue = wrkReportServiceApi.lokue_processor.q
     const aggregatorQueue = wrkReportServiceApi.lokue_aggregator.q
 
-    processorQueue.once('error:base', (err) => {
+    processorQueue.once(QUEUE_EVENT_NAMES.ERROR_BASE, (err) => {
       _emitError('processor', err)
       _stop()
     })
-    processorQueue.on('completed', (res) => {
+    processorQueue.on(QUEUE_EVENT_NAMES.COMPLETED, (res) => {
       _emitRes('processor', res)
     })
-    aggregatorQueue.once('error:base', (err) => {
+    aggregatorQueue.once(QUEUE_EVENT_NAMES.ERROR_BASE, (err) => {
       _emitError('aggregator', err)
       _stop()
     })
-    aggregatorQueue.on('completed', (res) => {
+    aggregatorQueue.on(QUEUE_EVENT_NAMES.COMPLETED, (res) => {
       _emitRes('aggregator', res)
     })
   } catch (err) {

workers/api.service.report.wrk.js

@@ -2,7 +2,7 @@
 
 const { WrkApi } = require('@bitfinex/bfx-wrk-api')
 const async = require('async')
-const path = require('path')
+const path = require('node:path')
 const argv = require('yargs')
   .option('dbId', {
     type: 'number',
@@ -50,6 +50,7 @@ const getI18next = require('./loc.api/i18next')
 const {
   PDFBufferUnderElectronCreationError
 } = require('./loc.api/errors')
+const QUEUE_EVENT_NAMES = require('./loc.api/queue/queue.event.names')
 
 class WrkReportServiceApi extends WrkApi {
   constructor (conf, ctx) {
@@ -219,18 +220,18 @@ class WrkReportServiceApi extends WrkApi {
     const processor = this.container.get(TYPES.Processor)
     const aggregator = this.container.get(TYPES.Aggregator)
 
-    processorQueue.on('job', processor)
-    aggregatorQueue.on('job', aggregator)
+    processorQueue.on(QUEUE_EVENT_NAMES.JOB, processor)
+    aggregatorQueue.on(QUEUE_EVENT_NAMES.JOB, aggregator)
 
-    processorQueue.on('error:base', (err) => {
+    processorQueue.on(QUEUE_EVENT_NAMES.ERROR_BASE, (err) => {
       // This error is intercepted and processed in the framework mode
       if (err instanceof PDFBufferUnderElectronCreationError) {
         return
       }
 
       this.logger.error('PROCESSOR:QUEUE:', err)
     })
-    aggregatorQueue.on('error:base', (err) => {
+    aggregatorQueue.on(QUEUE_EVENT_NAMES.ERROR_BASE, (err) => {
       this.logger.error('AGGREGATOR:QUEUE:', err)
     })
   }

workers/loc.api/generate-report-file/csv-writer/weighted-averages-report-csv-writer.js

@@ -3,6 +3,7 @@
 const {
   write
 } = require('../../queue/write-data-to-stream/helpers')
+const QUEUE_EVENT_NAMES = require('../../queue/queue.event.names')
 const {
   omitExtraParamFieldsForReportExport
 } = require('../helpers')
@@ -24,7 +25,7 @@ module.exports = (
   } = jobData ?? {}
   const { params } = args ?? {}
 
-  queue.emit('progress', 0)
+  queue.emit(QUEUE_EVENT_NAMES.PROGRESS, 0)
 
   if (typeof jobData === 'string') {
     await streamWriter(
@@ -35,7 +36,7 @@ module.exports = (
       }]
     )
 
-    queue.emit('progress', 100)
+    queue.emit(QUEUE_EVENT_NAMES.PROGRESS, 100)
 
     return
   }
@@ -79,5 +80,5 @@ module.exports = (
     ]
   )
 
-  queue.emit('progress', 100)
+  queue.emit(QUEUE_EVENT_NAMES.PROGRESS, 100)
 }

workers/loc.api/queue/aggregator.js

@@ -1,13 +1,11 @@
 'use strict'
 
-const { promisify } = require('util')
-const fs = require('fs')
-
-const unlink = promisify(fs.unlink)
+const { unlink } = require('node:fs/promises')
 
 const {
   moveFileToLocalStorage
 } = require('./helpers')
+const QUEUE_EVENT_NAMES = require('./queue.event.names')
 
 module.exports = (
   { isAddedUniqueEndingToReportFileName },
@@ -75,7 +73,7 @@ module.exports = (
         }
 
         job.done()
-        aggregatorQueue.emit('completed', {
+        aggregatorQueue.emit(QUEUE_EVENT_NAMES.COMPLETED, {
           newFilePaths,
           reportFilesMetadata,
           userInfo
@@ -107,28 +105,28 @@ module.exports = (
       }
 
       job.done()
-      aggregatorQueue.emit('completed', {
+      aggregatorQueue.emit(QUEUE_EVENT_NAMES.COMPLETED, {
         newFilePaths,
         reportFilesMetadata,
         userInfo
       })
     } catch (err) {
       if (err.syscall === 'unlink') {
-        aggregatorQueue.emit('error:unlink', job)
+        aggregatorQueue.emit(QUEUE_EVENT_NAMES.ERROR_UNLINK, job)
         job.done()
       } else {
         try {
           for (const filePath of job.data.filePaths) {
             await unlink(filePath)
           }
         } catch (e) {
-          aggregatorQueue.emit('error:unlink', job)
+          aggregatorQueue.emit(QUEUE_EVENT_NAMES.ERROR_UNLINK, job)
         }
 
         job.done(err)
       }
 
-      aggregatorQueue.emit('error:base', err, job)
+      aggregatorQueue.emit(QUEUE_EVENT_NAMES.ERROR_BASE, err, job)
     } finally {
       for (const stream of streamSet) {
         stream.destroy()

workers/loc.api/queue/processor.js

@@ -22,6 +22,7 @@ const {
 const TRANSLATION_NAMESPACES = require(
   '../i18next/translation.namespaces'
 )
+const QUEUE_EVENT_NAMES = require('./queue.event.names')
 
 const processReportFile = async (deps, args) => {
   const {
@@ -107,13 +108,13 @@ module.exports = (
   pdfWriter,
   i18next
 ) => {
-  processorQueue.on('completed', (result) => {
+  processorQueue.on(QUEUE_EVENT_NAMES.COMPLETED, (result) => {
     aggregatorQueue.addJob({
       ...result,
       s3Conf: conf.s3Conf
     })
   })
-  processorQueue.on('error:auth', (job) => {
+  processorQueue.on(QUEUE_EVENT_NAMES.ERROR_AUTH, (job) => {
     const data = cloneDeep(job.data)
     delete data.columnsCsv
 
@@ -192,7 +193,7 @@ module.exports = (
       }
 
       job.done()
-      processorQueue.emit('completed', {
+      processorQueue.emit(QUEUE_EVENT_NAMES.COMPLETED, {
         chunkCommonFolders,
         userInfo,
         userId,
@@ -209,18 +210,18 @@ module.exports = (
           await unlink(filePath)
         }
       } catch (err) {
-        processorQueue.emit('error:unlink', job)
+        processorQueue.emit(QUEUE_EVENT_NAMES.ERROR_UNLINK, job)
       }
 
       if (isAuthError(err)) {
         job.done()
-        processorQueue.emit('error:auth', job)
+        processorQueue.emit(QUEUE_EVENT_NAMES.ERROR_AUTH, job)
 
         return
       }
 
       job.done(err)
-      processorQueue.emit('error:base', err, job)
+      processorQueue.emit(QUEUE_EVENT_NAMES.ERROR_BASE, err, job)
     } finally {
       for (const stream of streamSet) {
         stream.destroy()

workers/loc.api/queue/queue.event.names.js

@@ -0,0 +1,10 @@
+'use strict'
+
+module.exports = {
+  ERROR_BASE: 'error:base',
+  ERROR_AUTH: 'error:auth',
+  ERROR_UNLINK: 'error:unlink',
+  COMPLETED: 'completed',
+  PROGRESS: 'progress',
+  JOB: 'job'
+}

workers/loc.api/queue/write-data-to-stream/helpers.js

@@ -3,6 +3,7 @@
 const { cloneDeep } = require('@bitfinex/lib-js-util-base')
 const moment = require('moment-timezone')
 
+const QUEUE_EVENT_NAMES = require('../queue.event.names')
 const dataNormalizer = require('./data-normalizer')
 
 const _validTxtTimeZone = (val, timezone, format) => {
@@ -197,11 +198,11 @@ const writeMessageToStream = (
   stream,
   message
 ) => {
-  processorQueue.emit('progress', 0)
+  processorQueue.emit(QUEUE_EVENT_NAMES.PROGRESS, 0)
 
   write([{ message }], stream)
 
-  processorQueue.emit('progress', 100)
+  processorQueue.emit(QUEUE_EVENT_NAMES.PROGRESS, 100)
 }
 
 const setDefaultParams = (args, method) => {
@@ -255,7 +256,7 @@ const progress = (
     ((currTime - start) / (end - start)) * 100
   )
 
-  queue.emit('progress', percent)
+  queue.emit(QUEUE_EVENT_NAMES.PROGRESS, percent)
 }
 
 module.exports = {