From d2dde4bfbc49e45c8e8b891ad120a298cab2bbef Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:01:47 +0300 Subject: [PATCH 01/11] Add queue event names file --- workers/loc.api/queue/queue.event.names.js | 10 ++++++++++ 1 file changed, 10 insertions(+) create mode 100644 workers/loc.api/queue/queue.event.names.js diff --git a/workers/loc.api/queue/queue.event.names.js b/workers/loc.api/queue/queue.event.names.js new file mode 100644 index 00000000..8754fc53 --- /dev/null +++ b/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' +} From 9375958ff3a1131fc564dc29a9faff8978b92fc7 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:04:59 +0300 Subject: [PATCH 02/11] Use queue event names in processor --- workers/loc.api/queue/processor.js | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/workers/loc.api/queue/processor.js b/workers/loc.api/queue/processor.js index ed44a73d..23d155bf 100644 --- a/workers/loc.api/queue/processor.js +++ b/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() From 794a3ae6da0a2054e50beac5eb941e4a8f7d0fc9 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:05:32 +0300 Subject: [PATCH 03/11] Use queue event names in aggregator --- workers/loc.api/queue/aggregator.js | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/workers/loc.api/queue/aggregator.js b/workers/loc.api/queue/aggregator.js index c942cf36..72d9b5b2 100644 --- a/workers/loc.api/queue/aggregator.js +++ b/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,14 +105,14 @@ 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 { @@ -122,13 +120,13 @@ module.exports = ( 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() From b9cdde98eadce7a7982eddedb8a3733becf648b6 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:06:32 +0300 Subject: [PATCH 04/11] Use queue event names in main service --- workers/api.service.report.wrk.js | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/workers/api.service.report.wrk.js b/workers/api.service.report.wrk.js index 5e5aff3b..45ca9047 100644 --- a/workers/api.service.report.wrk.js +++ b/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,10 +220,10 @@ 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 @@ -230,7 +231,7 @@ class WrkReportServiceApi extends WrkApi { 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) }) } From c55838183541cc4db29244521b443af5d241c43c Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:09:06 +0300 Subject: [PATCH 05/11] Use queue event names in helper for writing to stream --- workers/loc.api/queue/write-data-to-stream/helpers.js | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/workers/loc.api/queue/write-data-to-stream/helpers.js b/workers/loc.api/queue/write-data-to-stream/helpers.js index bb421a3b..0e9cc9ca 100644 --- a/workers/loc.api/queue/write-data-to-stream/helpers.js +++ b/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 = { From dcd87d10390b367c08eb4ca56b97ccb753afac65 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:09:31 +0300 Subject: [PATCH 06/11] Use queue event names in stream writer --- workers/loc.api/queue/write-data-to-stream/index.js | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/workers/loc.api/queue/write-data-to-stream/index.js b/workers/loc.api/queue/write-data-to-stream/index.js index 4ca0a945..dc4ba08d 100644 --- a/workers/loc.api/queue/write-data-to-stream/index.js +++ b/workers/loc.api/queue/write-data-to-stream/index.js @@ -5,6 +5,7 @@ const { cloneDeep } = require('@bitfinex/lib-js-util-base') const { omitExtraParamFieldsForReportExport } = require('../../generate-report-file/helpers') +const QUEUE_EVENT_NAMES = require('../queue.event.names') const { writeMessageToStream, setDefaultParams, @@ -69,7 +70,7 @@ module.exports = ( return accum }, {}) - processorQueue.emit('progress', 0) + processorQueue.emit(QUEUE_EVENT_NAMES.PROGRESS, 0) let count = 0 let serialRequestsCount = 0 @@ -127,7 +128,7 @@ module.exports = ( { ...jobData?.args?.params, symbols }, method ) - processorQueue.emit('progress', 100) + processorQueue.emit(QUEUE_EVENT_NAMES.PROGRESS, 100) break } @@ -136,7 +137,9 @@ module.exports = ( !Array.isArray(res) || res.length === 0 ) { - if (count > 0) processorQueue.emit('progress', 100) + if (count > 0) { + processorQueue.emit(QUEUE_EVENT_NAMES.PROGRESS, 100) + } break } @@ -176,7 +179,7 @@ module.exports = ( !Number.isInteger(currTime) || !Number.isInteger(nextPage) ) { - processorQueue.emit('progress', 100) + processorQueue.emit(QUEUE_EVENT_NAMES.PROGRESS, 100) break } From dc8300aa1a32edf625726f024c1c06277bc23336 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:11:16 +0300 Subject: [PATCH 07/11] Use queue event names in weighted averages writer --- .../csv-writer/weighted-averages-report-csv-writer.js | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/workers/loc.api/generate-report-file/csv-writer/weighted-averages-report-csv-writer.js b/workers/loc.api/generate-report-file/csv-writer/weighted-averages-report-csv-writer.js index f5411b4c..c62dbe8f 100644 --- a/workers/loc.api/generate-report-file/csv-writer/weighted-averages-report-csv-writer.js +++ b/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) } From 4b2000c8562c9ab86b2824261fcc9610815c6753 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:13:28 +0300 Subject: [PATCH 08/11] Use queue event names in test core helper --- test/helpers/helpers.core.js | 29 ++++++++++++++++------------- 1 file changed, 16 insertions(+), 13 deletions(-) diff --git a/test/helpers/helpers.core.js b/test/helpers/helpers.core.js index 12348ede..0d53ada3 100644 --- a/test/helpers/helpers.core.js +++ b/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,8 +106,8 @@ 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() @@ -112,8 +115,8 @@ const queuesToPromiseMulti = (queues, count, cb = () => { }) => { } 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) }) }) } From a94e78d83ed6ab87e723b8f9c291d13a94a7d121 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Thu, 4 Jun 2026 12:13:49 +0300 Subject: [PATCH 09/11] Use queue event names in test helper for loading --- test/helpers/helpers.queue-load.js | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/test/helpers/helpers.queue-load.js b/test/helpers/helpers.queue-load.js index 0155409a..2cae23f8 100644 --- a/test/helpers/helpers.queue-load.js +++ b/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) { From ef5eeb6f8d94b7af20afa7a63bff686fdd351f47 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Mon, 15 Jun 2026 12:08:48 +0300 Subject: [PATCH 10/11] Allow mocking of non-array data --- test/helpers/helpers.mock-rest-v2.js | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test/helpers/helpers.mock-rest-v2.js b/test/helpers/helpers.mock-rest-v2.js index f80f4c26..7c1ae355 100644 --- a/test/helpers/helpers.mock-rest-v2.js +++ b/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 } From 24daedef01c12976ceba3d01287ee8e29d799294 Mon Sep 17 00:00:00 2001 From: Vladimir Voronkov Date: Wed, 17 Jun 2026 10:10:14 +0300 Subject: [PATCH 11/11] Bump version up to 5.2.2 --- package-lock.json | 4 ++-- package.json | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/package-lock.json b/package-lock.json index a944efa1..d3b12d18 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@bitfinex/bfx-report", - "version": "5.2.1", + "version": "5.2.2", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@bitfinex/bfx-report", - "version": "5.2.1", + "version": "5.2.2", "license": "Apache-2.0", "dependencies": { "@bitfinex/bfx-facs-deflate": "git+https://github.com/bitfinexcom/bfx-facs-deflate.git", diff --git a/package.json b/package.json index bbfac92c..4f704a88 100644 --- a/package.json +++ b/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": {