refactor: query stuff to helpers

Boris · Boris · commit b22598b74c34 · 2020-04-13T16:54:09.000+06:00
diff --git a/example/src/schema/mutation/helpers/queueFind.ts b/example/src/schema/mutation/helpers/queueFind.ts
@@ -2,12 +2,19 @@ import { MutationError, ErrorCodeEnum } from './Error';
 import { createBullConnection } from '../../../connectRedis';
 import { Queue } from 'bullmq';
 
-export async function findQueue(prefix: string, queueName: string): Promise<Queue> {
+export async function findQueue(
+  prefix: string,
+  queueName: string,
+  checkExistence: boolean = true
+): Promise<Queue> {
   const connection = createBullConnection('custom');
-  const queueExists = await connection.exists([prefix, queueName, 'meta'].join(':'));
 
-  if (!queueExists) {
-    throw new MutationError('Queue not found!', ErrorCodeEnum.QUEUE_NOT_FOUND);
+  if (checkExistence) {
+    const queueExists = await connection.exists([prefix, queueName, 'meta'].join(':'));
+
+    if (!queueExists) {
+      throw new MutationError('Queue not found!', ErrorCodeEnum.QUEUE_NOT_FOUND);
+    }
   }
 
   const queue = new Queue(queueName, {
diff --git a/example/src/schema/mutation/helpers/queueGet.ts b/example/src/schema/mutation/helpers/queueGet.ts
diff --git a/example/src/schema/mutation/jobAdd.ts b/example/src/schema/mutation/jobAdd.ts
@@ -1,6 +1,6 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createJobAddFC(
   sc: SchemaComposer<any>
@@ -37,7 +37,7 @@ export function createJobAddFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.add(jobName, data, options);
       return {
         job,
diff --git a/example/src/schema/mutation/jobAddBulk.ts b/example/src/schema/mutation/jobAddBulk.ts
@@ -1,4 +1,4 @@
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 
@@ -44,7 +44,7 @@ export function createJobAddBulkFC(
         .getTypePlural(),
     },
     resolve: async (_, { prefix, queueName, jobs }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const jobsRes = await queue.addBulk(jobs);
       return {
         jobs: jobsRes,
diff --git a/example/src/schema/mutation/jobAddCron.ts b/example/src/schema/mutation/jobAddCron.ts
@@ -1,4 +1,4 @@
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 
@@ -49,7 +49,7 @@ export function createJobAddCronFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.add(jobName, data, options);
       return {
         job,
diff --git a/example/src/schema/mutation/jobAddEvery.ts b/example/src/schema/mutation/jobAddEvery.ts
@@ -1,4 +1,4 @@
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 
@@ -48,7 +48,7 @@ export function createJobAddEveryFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.add(jobName, data, options);
       return {
         job,
diff --git a/example/src/schema/mutation/jobDiscard.ts b/example/src/schema/mutation/jobDiscard.ts
@@ -1,7 +1,7 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { MutationError, ErrorCodeEnum } from './helpers/Error';
 import { getJobStatusEnumTC } from '../types';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createJobDiscardFC(
   sc: SchemaComposer<any>
@@ -23,7 +23,7 @@ export function createJobDiscardFC(
       id: 'String!',
     },
     resolve: async (_, { prefix, queueName, id }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.getJob(id);
       if (!job) throw new MutationError('Job not found!', ErrorCodeEnum.JOB_NOT_FOUND);
       await job.discard();
diff --git a/example/src/schema/mutation/jobLogAdd.ts b/example/src/schema/mutation/jobLogAdd.ts
@@ -1,7 +1,7 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { MutationError, ErrorCodeEnum } from './helpers/Error';
 import { getJobStatusEnumTC } from '../types';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createJobLogAddFC(
   sc: SchemaComposer<any>
@@ -24,7 +24,7 @@ export function createJobLogAddFC(
       row: 'String!',
     },
     resolve: async (_, { prefix, queueName, id, row }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.getJob(id);
       if (!job) throw new MutationError('Job not found!', ErrorCodeEnum.JOB_NOT_FOUND);
       const logRes = await job.log(row);
diff --git a/example/src/schema/mutation/jobMoveToCompleted.ts b/example/src/schema/mutation/jobMoveToCompleted.ts
@@ -1,4 +1,4 @@
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 
@@ -22,7 +22,7 @@ export function jobMoveToCompletedFC(
       id: 'String!',
     },
     resolve: async (_, { prefix, queueName, id }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.getJob(id);
       if (job) {
         await job.moveToCompleted({}, 'tokenmustbehere'); //TODO: нати где брать токен
diff --git a/example/src/schema/mutation/jobPromote.ts b/example/src/schema/mutation/jobPromote.ts
@@ -1,7 +1,7 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { MutationError, ErrorCodeEnum } from './helpers/Error';
 import { getJobStatusEnumTC } from '../types';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createjobPromoteFC(
   sc: SchemaComposer<any>
@@ -23,7 +23,7 @@ export function createjobPromoteFC(
       id: 'String!',
     },
     resolve: async (_, { prefix, queueName, id }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.getJob(id);
       if (!job) throw new MutationError('Job not found!', ErrorCodeEnum.JOB_NOT_FOUND);
       await job.promote();
diff --git a/example/src/schema/mutation/jobRemove.ts b/example/src/schema/mutation/jobRemove.ts
@@ -1,4 +1,4 @@
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 
@@ -22,7 +22,7 @@ export function createJobRremoveFC(
       id: 'String!',
     },
     resolve: async (_, { prefix, queueName, id }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.getJob(id);
       if (job) {
         await job.remove();
diff --git a/example/src/schema/mutation/jobRetry.ts b/example/src/schema/mutation/jobRetry.ts
@@ -1,7 +1,7 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { MutationError, ErrorCodeEnum } from './helpers/Error';
 import { JobStatusEnum, getJobStatusEnumTC } from '../types';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createJobRetryFC(
   sc: SchemaComposer<any>
@@ -23,7 +23,7 @@ export function createJobRetryFC(
       id: 'String!',
     },
     resolve: async (_, { prefix, queueName, id }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const job = await queue.getJob(id);
       if (!job) throw new MutationError('Job not found!', ErrorCodeEnum.JOB_NOT_FOUND);
       await job.retry();
diff --git a/example/src/schema/mutation/jobUpdate.ts b/example/src/schema/mutation/jobUpdate.ts
@@ -1,5 +1,5 @@
 import { MutationError, ErrorCodeEnum } from './helpers/Error';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 
@@ -23,7 +23,7 @@ export function createJobUpdateFC(
       data: 'JSON!',
     },
     resolve: async (_, { prefix, queueName, id, data }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       let job = await queue.getJob(id);
       if (!job) throw new MutationError('Job not found!', ErrorCodeEnum.JOB_NOT_FOUND);
       await job.update(data); //Данные заменяются полностью
diff --git a/example/src/schema/mutation/queueClean.ts b/example/src/schema/mutation/queueClean.ts
@@ -1,5 +1,5 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 import { getJobStatusEnumTC } from '../types';
 
 export function createQueueCleanFC(
@@ -36,7 +36,7 @@ export function createQueueCleanFC(
       }),
     },
     resolve: async (_, { prefix, queueName, filter: { grace, status, limit } }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       const jobsId = await queue.clean(grace, limit, status);
       return {
         jobsId,
diff --git a/example/src/schema/mutation/queueDrain.ts b/example/src/schema/mutation/queueDrain.ts
@@ -1,5 +1,5 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createQueueDrainFC(
   sc: SchemaComposer<any>
@@ -22,7 +22,7 @@ export function createQueueDrainFC(
       },
     },
     resolve: async (_, { prefix, queueName, delayed }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       await queue.drain(delayed);
       return {};
     },
diff --git a/example/src/schema/mutation/queuePause.ts b/example/src/schema/mutation/queuePause.ts
@@ -1,5 +1,5 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createQueuePauseFC(
   sc: SchemaComposer<any>
@@ -16,7 +16,7 @@ export function createQueuePauseFC(
       queueName: 'String!',
     },
     resolve: async (_, { prefix, queueName }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       await queue.pause();
       return {};
     },
diff --git a/example/src/schema/mutation/queueRemoveRepeatable.ts b/example/src/schema/mutation/queueRemoveRepeatable.ts
@@ -1,5 +1,5 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createRemoveRepeatableFC(
   sc: SchemaComposer<any>
@@ -20,7 +20,7 @@ export function createRemoveRepeatableFC(
       key: 'String!',
     },
     resolve: async (_, { prefix, queueName, key }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       await queue.removeRepeatableByKey(key);
       return {};
     },
diff --git a/example/src/schema/mutation/queueResume.ts b/example/src/schema/mutation/queueResume.ts
@@ -1,5 +1,5 @@
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
-import { getQueue } from './helpers/queueGet';
+import { findQueue } from './helpers/queueFind';
 
 export function createQueueResumeFC(
   sc: SchemaComposer<any>
@@ -16,7 +16,7 @@ export function createQueueResumeFC(
       queueName: 'String!',
     },
     resolve: async (_, { prefix, queueName }) => {
-      const queue = getQueue(prefix, queueName);
+      const queue = await findQueue(prefix, queueName);
       await queue.resume();
       return {};
     },
diff --git a/example/src/schema/query/_helpers.ts b/example/src/schema/query/_helpers.ts
@@ -0,0 +1,45 @@
+import { createBullConnection } from '../../connectRedis';
+import { Queue } from 'bullmq';
+
+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;
+}
+
+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],
+    };
+  });
+}
+
+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;
+}
diff --git a/example/src/schema/query/job.ts b/example/src/schema/query/job.ts
@@ -1,7 +1,6 @@
 import { SchemaComposer } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
-import { Queue } from 'bullmq';
-import { createBullConnection } from '../../connectRedis';
+import { getQueue } from './_helpers';
 
 export function createJobFC(sc: SchemaComposer<any>) {
   return {
@@ -15,10 +14,7 @@ export function createJobFC(sc: SchemaComposer<any>) {
       id: 'String!',
     },
     resolve: async (_, { prefix, queueName, id }) => {
-      const queue = new Queue(queueName, {
-        prefix,
-        connection: createBullConnection('queue'),
-      });
+      const queue = getQueue(prefix, queueName);
       if (!queue) return null;
       const job = await queue.getJob(id);
       if (!job) return null;
diff --git a/example/src/schema/query/queue.ts b/example/src/schema/query/queue.ts
@@ -1,7 +1,6 @@
 import { SchemaComposer } from 'graphql-compose';
 import { getQueueTC } from '../types/queue/Queue';
-import { Queue } from 'bullmq';
-import { createBullConnection } from '../../connectRedis';
+import { getQueue } from './_helpers';
 
 export function createQueueFC(sc: SchemaComposer<any>) {
   return {
@@ -14,12 +13,7 @@ export function createQueueFC(sc: SchemaComposer<any>) {
       queueName: 'String!',
     },
     resolve: async (_, { prefix, queueName }) => {
-      const queue = new Queue(queueName, {
-        prefix,
-        connection: createBullConnection('queue'),
-      });
-
-      return queue;
+      return getQueue(prefix, queueName);
     },
   };
 }
diff --git a/example/src/schema/query/queueKeys.ts b/example/src/schema/query/queueKeys.ts
diff --git a/example/src/schema/query/queues.ts b/example/src/schema/query/queues.ts
