feat: predefine queue for schema

Author: Boris

example/src/schema.ts

@@ -5,6 +5,10 @@ const { queryFields, mutationFields } = composeBull({
   schemaComposer,
   typePrefix: 'Prefix',
   jobDataTC: `type MyJobData { fieldA: String! fieldB: String}`,
+  queue: {
+    name: 'fetch_metrics',
+    prefix: 'bull.demo',
+  },
 });
 
 schemaComposer.Query.addFields({

src/composeBull.ts

@@ -19,7 +19,7 @@ import {
   createJobUpdateFC,
   createJobLogAddFC,
 } from './mutation';
-import { createMutationFC } from './helpers/wrapMutationFC';
+import { createMutationFC, predefineQueueArgs } from './helpers';
 
 export function composeBull(opts: Options & { schemaComposer?: SchemaComposer<any> }) {
   const sc = opts?.schemaComposer || schemaComposer;
@@ -28,27 +28,36 @@ export function composeBull(opts: Options & { schemaComposer?: SchemaComposer<an
     QueueTC: getQueueTC(sc, opts),
     JobTC: getJobTC(sc, opts),
     queryFields: {
-      queueKeys: createQueueKeysFC(sc, opts),
-      queues: createQueuesFC(sc, opts),
-      queue: createQueueFC(sc, opts),
-      job: createJobFC(sc, opts),
+      queueKeys: predefineQueueArgs(createQueueKeysFC(sc, opts), opts),
+      queues: predefineQueueArgs(createQueuesFC(sc, opts), opts),
+      queue: predefineQueueArgs(createQueueFC(sc, opts), opts),
+      job: predefineQueueArgs(createJobFC(sc, opts), opts),
     },
     mutationFields: {
-      queueClean: createMutationFC(createQueueCleanFC, sc, opts),
-      queueDrain: createMutationFC(createQueueDrainFC, sc, opts),
-      queuePause: createMutationFC(createQueuePauseFC, sc, opts),
-      queueResume: createMutationFC(createQueueResumeFC, sc, opts),
-      queueRemoveRepeatable: createMutationFC(createRemoveRepeatableFC, sc, opts),
-      jobAdd: createMutationFC(createJobAddFC, sc, opts),
-      jobAddBulk: createMutationFC(createJobAddBulkFC, sc, opts),
-      jobAddRepeatableCron: createMutationFC(createJobAddCronFC, sc, opts),
-      jobAddRepeatableEvery: createMutationFC(createJobAddEveryFC, sc, opts),
-      jobDiscard: createMutationFC(createJobDiscardFC, sc, opts),
-      jobPromote: createMutationFC(createjobPromoteFC, sc, opts),
-      jobRemove: createMutationFC(createJobRremoveFC, sc, opts),
-      jobRetry: createMutationFC(createJobRetryFC, sc, opts),
-      jobUpdate: createMutationFC(createJobUpdateFC, sc, opts),
-      jobLogAdd: createMutationFC(createJobLogAddFC, sc, opts),
+      queueClean: predefineQueueArgs(createMutationFC(createQueueCleanFC, sc, opts), opts),
+      queueDrain: predefineQueueArgs(createMutationFC(createQueueDrainFC, sc, opts), opts),
+      queuePause: predefineQueueArgs(createMutationFC(createQueuePauseFC, sc, opts), opts),
+      queueResume: predefineQueueArgs(createMutationFC(createQueueResumeFC, sc, opts), opts),
+      queueRemoveRepeatable: predefineQueueArgs(
+        createMutationFC(createRemoveRepeatableFC, sc, opts),
+        opts
+      ),
+      jobAdd: predefineQueueArgs(createMutationFC(createJobAddFC, sc, opts), opts),
+      jobAddBulk: predefineQueueArgs(createMutationFC(createJobAddBulkFC, sc, opts), opts),
+      jobAddRepeatableCron: predefineQueueArgs(
+        createMutationFC(createJobAddCronFC, sc, opts),
+        opts
+      ),
+      jobAddRepeatableEvery: predefineQueueArgs(
+        createMutationFC(createJobAddEveryFC, sc, opts),
+        opts
+      ),
+      jobDiscard: predefineQueueArgs(createMutationFC(createJobDiscardFC, sc, opts), opts),
+      jobPromote: predefineQueueArgs(createMutationFC(createjobPromoteFC, sc, opts), opts),
+      jobRemove: predefineQueueArgs(createMutationFC(createJobRremoveFC, sc, opts), opts),
+      jobRetry: predefineQueueArgs(createMutationFC(createJobRetryFC, sc, opts), opts),
+      jobUpdate: predefineQueueArgs(createMutationFC(createJobUpdateFC, sc, opts), opts),
+      jobLogAdd: predefineQueueArgs(createMutationFC(createJobLogAddFC, sc, opts), opts),
     },
   };
 }

src/helpers/Error.ts renamed to src/helpers/MutationError.ts

@@ -1,6 +1,20 @@
-import { MutationError, ErrorCodeEnum } from './Error';
-import { createBullConnection } from '../connectRedis';
+import { Title } from './queueTitles';
 import { Queue } from 'bullmq';
+import { createBullConnection } from '../connectRedis';
+import { MutationError, ErrorCodeEnum } from './MutationError';
+
+export function getQueues(titles: Array<Title>): Array<Queue> {
+  return titles.map((title) => getQueue(title.prefix, title.queueName));
+}
+
+export function getQueue(prefix: string, queueName: string): Queue {
+  const queue = new Queue(queueName, {
+    prefix,
+    connection: createBullConnection('queue'),
+  });
+
+  return queue;
+}
 
 export async function findQueue(
   prefix: string,

src/helpers/queueFind.ts renamed to src/helpers/gettingQueues.ts

@@ -0,0 +1,6 @@
+export * from './gettingQueues';
+export * from './queueTitles';
+export * from './MutationError';
+export * from './normalizePrefixGlob';
+export * from './wrapMutationFC';
+export * from './predefineQueueArgs';

src/helpers/index.ts

@@ -0,0 +1,14 @@
+export function normalizePrefixGlob(prefixGlob: string): string {
+  let prefixGlobNorm = prefixGlob;
+  const nameCase = prefixGlobNorm.split(':').length - 1;
+  if (nameCase >= 2) {
+    prefixGlobNorm = prefixGlobNorm.split(':').slice(0, 2).join(':') + ':';
+  } else if (nameCase === 1) {
+    prefixGlobNorm += prefixGlobNorm.endsWith(':') ? '*:' : ':';
+  } else {
+    prefixGlobNorm += ':*:';
+  }
+  prefixGlobNorm += 'meta';
+
+  return prefixGlobNorm;
+}

src/helpers/normalizePrefixGlob.ts

@@ -0,0 +1,30 @@
+import { Options } from './../definitions';
+import { ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
+
+export function predefineQueueArgs(
+  fieldConfig: ObjectTypeComposerFieldConfigAsObjectDefinition<any, any>,
+  opts: Options
+): ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> {
+  if (!opts.queue) {
+    fieldConfig.args = {
+      queueName: 'String!',
+      prefix: {
+        type: 'String',
+        defaultValue: 'bull',
+      },
+      ...fieldConfig.args,
+    };
+  } else {
+    const subResolve = fieldConfig.resolve || (() => ({}));
+    fieldConfig.resolve = async (source, args, context, info) => {
+      return subResolve(
+        source,
+        { queueName: opts.queue?.name, prefix: opts.queue?.prefix, ...args },
+        context,
+        info
+      );
+    };
+  }
+
+  return fieldConfig;
+}

src/helpers/predefineQueueArgs.ts

@@ -0,0 +1,17 @@
+import { createBullConnection } from '../connectRedis';
+import { normalizePrefixGlob } from './normalizePrefixGlob';
+
+export type Title = { prefix: string; queueName: string };
+
+export async function fetchQueueTitles(prefix: string): Promise<Array<Title>> {
+  const connection = createBullConnection('custom');
+  const keys = await connection.keys(normalizePrefixGlob(prefix));
+
+  return keys.map((key) => {
+    const parts = key.split(':');
+    return {
+      prefix: parts[0],
+      queueName: parts[1],
+    };
+  });
+}

src/helpers/queueTitles.ts

@@ -6,7 +6,7 @@ import {
   ObjectTypeComposer,
   inspect,
 } from 'graphql-compose';
-import { MutationError, ErrorCodeEnum } from './Error';
+import { MutationError, ErrorCodeEnum } from './MutationError';
 
 export enum MutationStatusEnum {
   OK = 'ok',

src/helpers/wrapMutationFC.ts

@@ -1,6 +1,6 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
-import { findQueue } from '../helpers/queueFind';
+import { findQueue } from '../helpers';
 import { Options } from '../definitions';
 import { createJobDataITC } from '../types/job/JobInput';
 
@@ -18,11 +18,6 @@ export function createJobAddFC(
       },
     }),
     args: {
-      prefix: {
-        type: 'String',
-        defaultValue: 'bull',
-      },
-      queueName: 'String!',
       jobName: 'String!',
       data: {
         type: () => createJobDataITC(sc, opts),