feat: add jobMoveToDelayed mutation

Author: Boris

src/composeBull.ts

@@ -18,6 +18,7 @@ import {
   createJobRetryFC,
   createJobUpdateFC,
   createJobLogAddFC,
+  createJobMoveToDelayedFC,
 } from './mutation';
 import { wrapMutationFC, wrapQueueArgs, composeFC } from './helpers';
 
@@ -52,6 +53,7 @@ export function composeBull(opts: Options & { schemaComposer?: SchemaComposer<an
       jobRetry: wrapMutation(createJobRetryFC),
       jobUpdate: wrapMutation(createJobUpdateFC),
       jobLogAdd: wrapMutation(createJobLogAddFC),
+      jobMoveToDelayed: wrapMutation(createJobMoveToDelayedFC),
     },
   };
 }

src/helpers/__tests__/normalizePrefixGlob-test.ts

@@ -1,6 +1,10 @@
 import { normalizePrefixGlob } from '../normalizePrefixGlob';
 
 describe('normalizePrefixGlob', () => {
+  it('check empty mask', () => {
+    expect(normalizePrefixGlob('')).toBe('*:*:meta');
+  });
+
   it('check simple mask', () => {
     expect(normalizePrefixGlob('bull*')).toBe('bull*:*:meta');
     expect(normalizePrefixGlob('bull')).toBe('bull:*:meta');
@@ -10,11 +14,13 @@ describe('normalizePrefixGlob', () => {
     expect(normalizePrefixGlob('bull:metrics*')).toBe('bull:metrics*:meta');
   });
 
-  it('check mask with queue name', () => {
+  it('check mask with queue name complete', () => {
     expect(normalizePrefixGlob('bull:metrics')).toBe('bull:metrics:meta');
+    expect(normalizePrefixGlob('bull*:metrics')).toBe('bull*:metrics:meta');
   });
 
   it('check mask with prefix which contains semicolons', () => {
     expect(normalizePrefixGlob('my:service:bull:metrics')).toBe('my:service:bull:metrics:meta');
+    expect(normalizePrefixGlob('my:service*:bull:metrics*')).toBe('my:service*:bull:metrics*:meta');
   });
 });

src/helpers/normalizePrefixGlob.ts

@@ -1,13 +1,15 @@
 export function normalizePrefixGlob(prefixGlob: string): string {
   let prefixGlobNorm = prefixGlob;
-  const nameCase = prefixGlobNorm.split(':');
-  if (nameCase.length >= 3) {
-    prefixGlobNorm = nameCase.filter((s) => s !== '').join(':') + ':';
-  } else if (nameCase.length === 2) {
+  const sectionsCount = prefixGlobNorm.split(':').length - 1;
+
+  if (sectionsCount > 1) {
+    prefixGlobNorm += prefixGlobNorm.endsWith(':') ? '' : ':';
+  } else if (sectionsCount == 1) {
     prefixGlobNorm += prefixGlobNorm.endsWith(':') ? '*:' : ':';
   } else {
-    prefixGlobNorm += ':*:';
+    prefixGlobNorm += prefixGlobNorm.trim().length > 0 ? ':*:' : '*:*:';
   }
+
   prefixGlobNorm += 'meta';
 
   return prefixGlobNorm;

src/mutation/index.ts

@@ -13,3 +13,4 @@ export { createJobRremoveFC } from './jobRemove';
 export { createJobRetryFC } from './jobRetry';
 export { createJobUpdateFC } from './jobUpdate';
 export { createJobLogAddFC } from './jobLogAdd';
+export { createJobMoveToDelayedFC } from './jobMoveToDelayed';

src/mutation/jobMoveToCompleted.ts

@@ -3,7 +3,7 @@ import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from
 import { getJobTC } from '../types/job/Job';
 import { Options } from '../definitions';
 
-export function jobMoveToCompletedFC(
+export function createJobMoveToCompletedFC(
   sc: SchemaComposer<any>,
   opts: Options
 ): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {

src/mutation/jobMoveToDelayed.ts

@@ -0,0 +1,46 @@
+import { MutationError, ErrorCodeEnum } from './../helpers/MutationError';
+import { findQueue } from '../helpers';
+import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+import { getJobTC } from '../types/job/Job';
+import { Options } from '../definitions';
+
+export function createJobMoveToDelayedFC(
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  const { typePrefix } = opts;
+
+  return {
+    description: 'Moves job from active to delayed.',
+    type: sc.createObjectTC({
+      name: `${typePrefix}JobMoveToDelayedPayload`,
+      fields: {
+        job: getJobTC(sc, opts),
+        willBeProcessedOn: 'Date',
+      },
+    }),
+    args: {
+      prefix: {
+        type: 'String!',
+        defaultValue: 'bull',
+      },
+      queueName: 'String!',
+      id: 'String!',
+      delay: {
+        type: 'Int!',
+        defaultValue: 60000,
+      },
+    },
+    resolve: async (_, { prefix, queueName, id, delay }) => {
+      const queue = await findQueue(prefix, queueName, opts);
+      const job = await queue.getJob(id);
+      if (!job) throw new MutationError('Job not found!', ErrorCodeEnum.JOB_NOT_FOUND);
+      const willBeProcessedOn = Date.now() + delay;
+      await job.moveToDelayed(willBeProcessedOn);
+      return {
+        willBeProcessedOn,
+        job,
+      };
+    },
+  };
+}