feat(queue): add get version

.github/workflows/release.yml

@@ -2,7 +2,7 @@ name: Release
 on:
   push:
     branches:
-      - master
+      - v3.x
 
 env:
   HUSKY: 0

package.json

@@ -167,6 +167,21 @@
     "url": "https://github.com/taskforcesh/bullmq.git"
   },
   "release": {
+    "branches": [
+      "master",
+      {
+        "name": "v5.x",
+        "range": "5.x"
+      },
+      {
+        "name": "v4.x",
+        "range": "4.x"
+      },
+      {
+        "name": "v3.x",
+        "range": "3.x"
+      }
+    ],
     "plugins": [
       [
         "@semantic-release/commit-analyzer",

src/classes/queue.ts

@@ -8,7 +8,7 @@ import {
   RepeatOptions,
 } from '../interfaces';
 import { FinishedStatus, JobsOptions, MinimalQueue } from '../types';
-import { isRedisInstance } from '../utils';
+import { isRedisInstance, readPackageJson } from '../utils';
 import { Job } from './job';
 import { QueueGetters } from './queue-getters';
 import { Repeat } from './repeat';
@@ -96,6 +96,7 @@ export class Queue<
 > extends QueueGetters<DataType, ResultType, NameType> {
   token = v4();
   jobsOpts: BaseJobOptions;
+  opts: QueueOptions;
   private _repeat?: Repeat;
 
   constructor(
@@ -117,12 +118,8 @@ export class Queue<
 
     this.waitUntilReady()
       .then(client => {
-        if (!this.closing) {
-          client.hset(
-            this.keys.meta,
-            'opts.maxLenEvents',
-            get(opts, 'streams.events.maxLen', 10000),
-          );
+        if (!this.closing && !opts?.skipMetasUpdate) {
+          return client.hmset(this.keys.meta, this.metaValues);
         }
       })
       .catch(err => {
@@ -169,6 +166,25 @@ export class Queue<
     return { ...this.jobsOpts };
   }
 
+  get metaValues(): Record<string, string | number> {
+    const { name, version } = readPackageJson();
+
+    return {
+      'opts.maxLenEvents': this.opts?.streams?.events?.maxLen ?? 10000,
+      version: `${name}:${version}`,
+    };
+  }
+
+  /**
+   * Get library version.
+   *
+   * @returns the content of the meta.library field.
+   */
+  async getVersion(): Promise<string> {
+    const client = await this.client;
+    return await client.hget(this.keys.meta, 'version');
+  }
+
   get repeat(): Promise<Repeat> {
     return new Promise<Repeat>(async resolve => {
       if (!this._repeat) {

src/interfaces/queue-options.ts

@@ -53,6 +53,16 @@ export interface QueueOptions extends QueueBaseOptions {
     };
   };
 
+  /**
+   * Skip Meta update.
+   *
+   * If true, the queue will not update the metadata of the queue.
+   * Useful for read-only systems that do should not update the metadata.
+   *
+   * @defaultValue false
+   */
+  skipMetasUpdate?: boolean;
+
   settings?: AdvancedRepeatOptions;
 }
 

src/utils.ts

@@ -4,6 +4,8 @@ import { Cluster, Redis } from 'ioredis';
 import { CONNECTION_CLOSED_ERROR_MSG } from 'ioredis/built/utils';
 import * as semver from 'semver';
 import { ChildMessage, RedisClient } from './interfaces';
+import { join } from 'path';
+import { readFileSync } from 'fs';
 
 export const errorObject: { [index: string]: any } = { value: null };
 
@@ -183,3 +185,24 @@ export const errorToJSON = (value: any): Record<string, any> => {
 export const WORKER_SUFFIX = '';
 
 export const QUEUE_EVENT_SUFFIX = ':qe';
+
+export const readPackageJson: () => { name: string; version: string } = () => {
+  const packageJsonPossiblePaths = [
+    join(__dirname, '../package.json'),
+    join(__dirname, '../../package.json'),
+    join(__dirname, '../../../package.json'),
+  ];
+
+  for (const path of packageJsonPossiblePaths) {
+    try {
+      return JSON.parse(readFileSync(path, 'utf-8'));
+    } catch (err) {
+      if ((<any>err).code === 'ENOENT') {
+        continue;
+      }
+      console.log(err);
+    }
+  }
+
+  return { name: 'bullmq', version: '0.0.0' };
+};

tests/test_queue.ts

@@ -78,6 +78,7 @@ import { delay, removeAllQueueData } from '../src/utils';
 
 describe('queues', function () {
   const sandbox = sinon.createSandbox();
+  const redisHost = process.env.REDIS_HOST || 'localhost';
 
   let queue: Queue;
   let queueName: string;
@@ -96,6 +97,23 @@ describe('queues', function () {
     await removeAllQueueData(new IORedis(), queueName);
   });
 
+  it('should return the queue version', async () => {
+    const queue = new Queue(queueName, { connection });
+    const version = await queue.getVersion();
+    const { version: pkgJsonVersion, name } = require('../package.json');
+    expect(version).to.be.equal(`${name}:${pkgJsonVersion}`);
+    return queue.close();
+  });
+
+  it('should return default library version when using skipMetasUpdate', async () => {
+    const exQueueName = `test-${v4()}`;
+    const queue = new Queue(exQueueName, { connection, skipMetasUpdate: true });
+    const version = await queue.getVersion();
+    expect(version).to.be.equal(null);
+    await queue.close();
+    await removeAllQueueData(new IORedis(redisHost), exQueueName);
+  });
+
   //TODO: restore this tests in next breaking change
   describe.skip('.add', () => {
     describe('when jobId is provided as integer', () => {