feat(queue): add version support

src/classes/queue.ts

@@ -13,6 +13,7 @@ import { Job } from './job';
 import { QueueGetters } from './queue-getters';
 import { Repeat } from './repeat';
 import { RedisConnection } from './redis-connection';
+import { readPackageJson } from '../utils';
 
 export interface ObliterateOpts {
   /**
@@ -96,6 +97,7 @@ export class Queue<
 > extends QueueGetters<DataType, ResultType, NameType> {
   token = v4();
   jobsOpts: BaseJobOptions;
+  opts: QueueOptions;
   private _repeat?: Repeat;
 
   constructor(
@@ -116,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 => {
@@ -168,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

@@ -54,6 +54,19 @@ 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;
+
+  /**
+   * Advanced options for the repeatable jobs.
+   */
   settings?: AdvancedRepeatOptions;
 }
 

src/utils.ts

@@ -9,6 +9,8 @@ import { CONNECTION_CLOSED_ERROR_MSG } from 'ioredis/built/utils';
 import { ChildMessage, RedisClient } from './interfaces';
 import { EventEmitter } from 'events';
 import * as semver from 'semver';
+import { join } from 'path';
+import { readFileSync } from 'fs';
 
 export const errorObject: { [index: string]: any } = { value: null };
 
@@ -215,3 +217,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

@@ -36,6 +36,23 @@ describe('queues', function () {
     await connection.quit();
   });
 
+  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', () => {