feat: Implement process worker with bullmq

packages/backend/.env-example

@@ -10,3 +10,5 @@ POSTGRES_USERNAME=automatish_development_user
 POSTGRES_PASSWORD=
 POSTGRES_ENABLE_SSL=false
 ENCRYPTION_KEY=sample-encryption-key
+REDIS_PORT=6379
+REDIS_HOST=127.0.0.1

packages/backend/package.json

@@ -4,6 +4,7 @@
   "description": "> TODO: description",
   "scripts": {
     "dev": "nodemon --watch 'src/**/*.ts' --exec 'ts-node' src/app.ts",
+    "worker": "nodemon --watch 'src/**/*.ts' --exec 'ts-node' src/worker.ts",
     "build": "tsc",
     "start": "node dist/src/app.js",
     "test": "ava",
@@ -21,6 +22,7 @@
     "ajv-formats": "^2.1.1",
     "axios": "0.24.0",
     "bcrypt": "^5.0.1",
+    "bullmq": "^1.76.1",
     "cors": "^2.8.5",
     "crypto-js": "^4.1.1",
     "debug": "~2.6.9",

packages/backend/src/config/app.ts

@@ -16,6 +16,8 @@ type AppConfig = {
   baseUrl?: string;
   encryptionKey: string;
   serveWebAppSeparately: boolean;
+  redisHost: string;
+  redisPort: number;
 };
 
 const appConfig: AppConfig = {
@@ -33,6 +35,8 @@ const appConfig: AppConfig = {
   encryptionKey: process.env.ENCRYPTION_KEY,
   serveWebAppSeparately:
     process.env.SERVE_WEB_APP_SEPARATELY === 'true' ? true : false,
+  redisHost: process.env.REDIS_HOST || '127.0.0.1',
+  redisPort: parseInt(process.env.REDIS_PORT) || 6379,
 };
 
 if (appConfig.serveWebAppSeparately) {

packages/backend/src/config/redis.ts

@@ -0,0 +1,8 @@
+import appConfig from './app';
+
+const redisConfig = {
+  host: appConfig.redisHost,
+  port: appConfig.redisPort,
+};
+
+export default redisConfig;

packages/backend/src/graphql/mutations/execute-flow.ts

@@ -2,6 +2,7 @@ import { GraphQLString, GraphQLNonNull } from 'graphql';
 import RequestWithCurrentUser from '../../types/express/request-with-current-user';
 import executeFlowType from '../types/execute-flow';
 import Processor from '../../services/processor';
+import processorQueue from '../../queues/processor';
 
 type Params = {
   stepId: string;
@@ -21,6 +22,12 @@ const executeFlowResolver = async (
   const flow = await step.$relatedQuery('flow');
   const data = await new Processor(flow, step).run();
 
+  // TODO: Use this snippet to execute flows with the background job.
+  // const data = processorQueue.add('processorJob', {
+  //   flowId: flow.id,
+  //   stepId: step.id,
+  // });
+
   await step.$query().patch({
     status: 'completed',
   });

packages/backend/src/queues/processor.ts

@@ -0,0 +1,8 @@
+import { Queue } from 'bullmq';
+import redisConfig from '../config/redis';
+
+const processorQueue = new Queue('processor', {
+  connection: redisConfig,
+});
+
+export default processorQueue;

packages/backend/src/worker.ts

@@ -0,0 +1,2 @@
+import './config/database';
+import './workers/processor';

packages/backend/src/workers/processor.ts

@@ -0,0 +1,31 @@
+import { Worker } from 'bullmq';
+import Processor from '../services/processor';
+import redisConfig from '../config/redis';
+import Step from '../models/step';
+import logger from '../helpers/logger';
+
+const worker = new Worker(
+  'processor',
+  async (job) => {
+    const step = await Step.query()
+      .withGraphFetched('connection')
+      .findOne({
+        'steps.id': job.data.stepId,
+      })
+      .throwIfNotFound();
+
+    const flow = await step.$relatedQuery('flow');
+
+    const data = await new Processor(flow, step).run();
+    return data;
+  },
+  { connection: redisConfig }
+);
+
+worker.on('completed', (job) => {
+  logger.info(`${job.id} has completed!`);
+});
+
+worker.on('failed', (job, err) => {
+  logger.info(`${job.id} has failed with ${err.message}`);
+});