feat: add queue delete mutation

Author: Boris Dorofeev

src/composeBull.ts

@@ -8,6 +8,7 @@ import {
   createQueuePauseFC,
   createQueueResumeFC,
   createRemoveRepeatableFC,
+  createQueueDeleteFC,
   createJobAddFC,
   createJobAddBulkFC,
   createJobAddCronFC,
@@ -43,6 +44,7 @@ export function composeBull(opts: Options & { schemaComposer?: SchemaComposer<an
       queuePause: wrapMutation(createQueuePauseFC),
       queueResume: wrapMutation(createQueueResumeFC),
       queueRemoveRepeatable: wrapMutation(createRemoveRepeatableFC),
+      queueDelete: wrapMutation(createQueueDeleteFC),
       jobAdd: wrapMutation(createJobAddFC),
       jobAddBulk: wrapMutation(createJobAddBulkFC),
       jobAddRepeatableCron: wrapMutation(createJobAddCronFC),

src/helpers/deleteKeys.ts

@@ -0,0 +1,43 @@
+import { MutationError, ErrorCodeEnum } from './MutationError';
+import { getBullConnection } from './getBullConnection';
+import { Options } from '../definitions';
+
+export async function deleteQueue(
+  prefix: string,
+  queueName: string,
+  opts: Options,
+  checkExistence: boolean = true
+): Promise<number> {
+  const connection = getBullConnection(opts);
+
+  const fullName = [prefix, queueName].join(':');
+
+  if (checkExistence) {
+    const queueExists = await connection.exists([fullName, 'meta'].join(':'));
+
+    if (!queueExists) {
+      throw new MutationError('Queue not found!', ErrorCodeEnum.QUEUE_NOT_FOUND);
+    }
+  }
+
+  //redis-cli: scan 0 match fullName* count 300
+  const stream = connection.scanStream({ match: fullName + '*', count: 300 });
+
+  let total = 0;
+
+  stream.on('data', async (keys) => {
+    for (let i = 0; i < keys.length; i++) {
+      const del = await connection.del(keys[i]);
+      if (del) {
+        console.log(keys[i]);
+        total++;
+      }
+    }
+  });
+
+  stream.on('end', function () {
+    console.log('all keys have been deleted!');
+  });
+
+  return total;
+}

src/helpers/index.ts

@@ -5,3 +5,4 @@ export * from './normalizePrefixGlob';
 export * from './wrapMutationFC';
 export * from './wrapQueueArgs';
 export * from './composeFC';
+export * from './deleteKeys';

src/mutation/index.ts

@@ -2,6 +2,7 @@ export { createQueueCleanFC } from './queueClean';
 export { createQueueDrainFC } from './queueDrain';
 export { createQueuePauseFC } from './queuePause';
 export { createQueueResumeFC } from './queueResume';
+export { createQueueDeleteFC } from './queueDelete';
 export { createRemoveRepeatableFC } from './queueRemoveRepeatable';
 export { createJobAddFC } from './jobAdd';
 export { createJobAddBulkFC } from './jobAddBulk';

src/mutation/queueDelete.ts

@@ -0,0 +1,34 @@
+import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+import { deleteQueue } from '../helpers';
+import { Options } from '../definitions';
+
+export function createQueueDeleteFC(
+  sc: SchemaComposer<any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  const { typePrefix } = opts;
+
+  return {
+    type: sc.createObjectTC({
+      name: `${typePrefix}QueueDeletePayload`,
+      fields: {
+        total: 'Int',
+      },
+    }),
+    args: {
+      prefix: {
+        type: 'String!',
+        defaultValue: 'bull',
+      },
+      queueName: 'String!',
+      checkExistence: {
+        type: 'Boolean',
+        defaultValue: true,
+      },
+    },
+    resolve: async (_, { prefix, queueName, checkExistence }) => {
+      const total = await deleteQueue(prefix, queueName, opts, checkExistence);
+      return { total };
+    },
+  };
+}