feat: limit job data size for all addJob mutations

relates: #83

Author: Boris Dorofeev

src/definitions.ts

@@ -11,4 +11,5 @@ export type Options = {
   };
   redis?: RedisOptions | Redis;
   redisEvents?: RedisOptions | Redis;
+  maxSizeOfJobData?: number /** in bytes */;
 };

src/helpers/roughSizeOfObject.ts

@@ -0,0 +1,36 @@
+import { Options } from '../definitions';
+
+function roughSizeOfObject(object: unknown): number {
+  const objectList = [];
+  const stack = [object];
+  let bytes = 0;
+
+  while (stack.length) {
+    const value = stack.pop();
+
+    if (typeof value === 'boolean') {
+      bytes += 4;
+    } else if (typeof value === 'string') {
+      bytes += value.length * 2;
+    } else if (typeof value === 'number') {
+      bytes += 8;
+    } else if (typeof value === 'object' && objectList.indexOf(value as never) === -1) {
+      objectList.push(value as never);
+
+      for (const i in value) {
+        stack.push(value[i]);
+      }
+    }
+  }
+  return bytes;
+}
+
+export const defaultMaxSizeOfObjectData = 10000;
+
+export function checkJobDataSize(opts: Options, data: unknown): void {
+  const allowableSize = opts?.maxSizeOfJobData || defaultMaxSizeOfObjectData;
+  const size = roughSizeOfObject(data);
+  if (size > allowableSize) {
+    throw new Error(`Job data ${size} exceeds allowable size ${allowableSize}`);
+  }
+}

src/mutation/jobAdd.ts

@@ -3,6 +3,7 @@ import { getJobTC } from '../types/job/Job';
 import { findQueue } from '../helpers';
 import { Options } from '../definitions';
 import { createJobDataITC } from '../types/job/JobInput';
+import { checkJobDataSize } from '../helpers/roughSizeOfObject';
 
 export function createJobAddFC(
   sc: SchemaComposer<any>,
@@ -44,6 +45,7 @@ export function createJobAddFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
+      checkJobDataSize(opts, data);
       const queue = await findQueue(prefix, queueName, opts);
       const job = await queue.add(jobName, data, options);
       return {

src/mutation/jobAddBulk.ts

@@ -2,6 +2,7 @@ import { findQueue } from '../helpers';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 import { Options } from '../definitions';
+import { checkJobDataSize } from '../helpers/roughSizeOfObject';
 
 export function createJobAddBulkFC(
   sc: SchemaComposer<any>,
@@ -46,6 +47,14 @@ export function createJobAddBulkFC(
       }).List,
     },
     resolve: async (_, { prefix, queueName, jobs }) => {
+      if (Array.isArray(jobs)) {
+        for (const job of jobs) {
+          checkJobDataSize(opts, job);
+        }
+      } else {
+        throw new Error('jobAddBulk: jobs argument must be an array');
+      }
+
       const queue = await findQueue(prefix, queueName, opts);
       const jobsRes = await queue.addBulk(jobs);
       return {

src/mutation/jobAddCron.ts

@@ -2,6 +2,7 @@ import { findQueue } from '../helpers';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 import { Options } from '../definitions';
+import { checkJobDataSize } from '../helpers/roughSizeOfObject';
 
 export function createJobAddCronFC(
   sc: SchemaComposer<any>,
@@ -54,6 +55,7 @@ export function createJobAddCronFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
+      checkJobDataSize(opts, data);
       const queue = await findQueue(prefix, queueName, opts);
       const job = await queue.add(jobName, data, options);
       return {

src/mutation/jobAddEvery.ts

@@ -2,6 +2,7 @@ import { findQueue } from '../helpers';
 import { SchemaComposer, ObjectTypeComposerFieldConfigAsObjectDefinition } from 'graphql-compose';
 import { getJobTC } from '../types/job/Job';
 import { Options } from '../definitions';
+import { checkJobDataSize } from '../helpers/roughSizeOfObject';
 
 export function createJobAddEveryFC(
   sc: SchemaComposer<any>,
@@ -53,6 +54,7 @@ export function createJobAddEveryFC(
       }),
     },
     resolve: async (_, { prefix, queueName, jobName, data, options }) => {
+      checkJobDataSize(opts, data);
       const queue = await findQueue(prefix, queueName, opts);
       const job = await queue.add(jobName, data, options);
       return {