feat(job-queue-plugin): Create BullMQJobQueuePlugin

This introduces a new plugin which implements a job queue based on Redis/BullMQ

docs/data/build.json

@@ -1,4 +1,4 @@
 {
-  "version": "1.1.0",
-  "commit": "e1ba5d6ca"
+  "version": "1.1.3",
+  "commit": "812b2cbbd"
 }

package.json

@@ -46,12 +46,12 @@
     "find": "^0.3.0",
     "graphql": "15.5.0",
     "husky": "^4.3.0",
-    "jest": "^26.6.3",
+    "jest": "^27.0.6",
     "klaw-sync": "^6.0.0",
     "lerna": "^4.0.0",
     "lint-staged": "^10.5.4",
     "prettier": "^2.2.1",
-    "ts-jest": "^26.5.3",
+    "ts-jest": "^27.0.4",
     "ts-node": "^9.1.1",
     "tslint": "^6.1.3",
     "typescript": "4.1.5"

packages/core/src/health-check/worker-health-indicator.ts

@@ -3,6 +3,7 @@ import { HealthCheckError, HealthIndicator, HealthIndicatorResult } from '@nestj
 
 import { ConfigService } from '../config/config.service';
 import { isInspectableJobQueueStrategy } from '../config/job-queue/inspectable-job-queue-strategy';
+import { Logger } from '../config/logger/vendure-logger';
 import { JobQueue } from '../job-queue/job-queue';
 import { JobQueueService } from '../job-queue/job-queue.service';
 
@@ -38,6 +39,7 @@ export class WorkerHealthIndicator extends HealthIndicator implements OnModuleIn
             try {
                 isHealthy = !!(await job.updates({ timeoutMs: 10000 }).toPromise());
             } catch (e) {
+                Logger.error(e.message);
                 isHealthy = false;
             }
             const result = this.getStatus('worker', isHealthy);

packages/dev-server/dev-config.ts

@@ -11,9 +11,12 @@ import {
     VendureConfig,
 } from '@vendure/core';
 import { defaultEmailHandlers, EmailPlugin } from '@vendure/email-plugin';
+import { BullMQJobQueuePlugin } from '@vendure/job-queue-plugin/package/bullmq';
 import path from 'path';
 import { ConnectionOptions } from 'typeorm';
 
+import { JobQueueTestPlugin } from './test-plugins/job-queue-test/job-queue-test-plugin';
+
 /**
  * Config settings used during development
  */
@@ -64,7 +67,9 @@ export const devConfig: VendureConfig = {
             assetUploadDir: path.join(__dirname, 'assets'),
         }),
         DefaultSearchPlugin,
-        DefaultJobQueuePlugin,
+        BullMQJobQueuePlugin.init({}),
+        // DefaultJobQueuePlugin,
+        // JobQueueTestPlugin.init({ queueCount: 10 }),
         // ElasticsearchPlugin.init({
         //     host: 'http://localhost',
         //     port: 9200,

packages/dev-server/index-worker.ts

@@ -2,12 +2,9 @@ import { bootstrapWorker } from '@vendure/core';
 
 import { devConfig } from './dev-config';
 
-// https://github.com/vendure-ecommerce/vendure/issues/152
-// fix race condition when modifying DB
-devConfig.dbConnectionOptions = { ...devConfig.dbConnectionOptions, synchronize: false };
-
 bootstrapWorker(devConfig)
     .then(worker => worker.startJobQueue())
+    // .then(worker => worker.startHealthCheckServer({ port: 3001 }))
     .catch(err => {
         // tslint:disable-next-line
         console.log(err);

packages/dev-server/index.ts

@@ -7,7 +7,7 @@ import { devConfig } from './dev-config';
  */
 bootstrap(devConfig)
     .then(app => {
-        if (process.env.RUN_JOB_QUEUE) {
+        if (process.env.RUN_JOB_QUEUE === '1') {
             app.get(JobQueueService).start();
         }
     })

packages/dev-server/test-plugins/job-queue-test/job-queue-test-plugin.ts

@@ -3,8 +3,8 @@ import { Args, Mutation, Resolver } from '@nestjs/graphql';
 import { JobState } from '@vendure/common/lib/generated-types';
 import { JobQueue, JobQueueService, Logger, PluginCommonModule, VendurePlugin } from '@vendure/core';
 import { gql } from 'apollo-server-core';
-import { of } from 'rxjs';
-import { catchError, map, tap } from 'rxjs/operators';
+import { forkJoin, Observable, of } from 'rxjs';
+import { catchError, map } from 'rxjs/operators';
 
 interface TaskConfigInput {
     intervalMs: number;
@@ -13,49 +13,65 @@ interface TaskConfigInput {
     subscribeToResult: boolean;
 }
 
+let queueCount = 1;
+
 @Injectable()
 export class JobQueueTestService implements OnModuleInit {
-    private myQueue: JobQueue<{ intervalMs: number; shouldFail: boolean }>;
+    private queues: Array<JobQueue<{ intervalMs: number; shouldFail: boolean }>> = [];
 
     constructor(private jobQueueService: JobQueueService) {}
 
     async onModuleInit() {
-        this.myQueue = await this.jobQueueService.createQueue({
-            name: 'my-queue',
-            process: async job => {
-                Logger.info(`Starting job ${job.id}, shouldFail: ${JSON.stringify(job.data.shouldFail)}`);
-                let progress = 0;
-                while (progress < 100) {
-                    // Logger.info(`Job ${job.id} progress: ${progress}`);
-                    await new Promise(resolve => setTimeout(resolve, job.data.intervalMs));
-                    progress += 10;
-                    job.setProgress(progress);
-                    if (progress > 70 && job.data.shouldFail) {
-                        Logger.warn(`Job ${job.id} will fail`);
-                        throw new Error(`Job failed!!`);
+        for (let i = 0; i < queueCount; i++) {
+            const queue: JobQueue<{
+                intervalMs: number;
+                shouldFail: boolean;
+            }> = await this.jobQueueService.createQueue({
+                name: `test-queue-${i + 1}`,
+                process: async job => {
+                    Logger.info(`Starting job ${job.id}, shouldFail: ${JSON.stringify(job.data.shouldFail)}`);
+                    let progress = 0;
+                    while (progress < 100) {
+                        // Logger.info(`Job ${job.id} progress: ${progress}`);
+                        await new Promise(resolve => setTimeout(resolve, job.data.intervalMs));
+                        progress += 10;
+                        job.setProgress(progress);
+                        if (progress > 70 && job.data.shouldFail) {
+                            Logger.warn(`Job ${job.id} will fail`);
+                            throw new Error(`Job failed!!`);
+                        }
                     }
-                }
-                Logger.info(`Completed job ${job.id}`);
-                return 'Done!';
-            },
-        });
+                    Logger.info(`Completed job ${job.id}`);
+                    return 'Done!';
+                },
+            });
+            this.queues.push(queue);
+        }
     }
 
     async startTask(input: TaskConfigInput) {
         const { intervalMs, shouldFail, subscribeToResult, retries } = input;
-        const job = await this.myQueue.add({ intervalMs, shouldFail }, { retries });
+        const updates: Array<Observable<number>> = [];
+        for (const queue of this.queues) {
+            const job = await queue.add({ intervalMs, shouldFail }, { retries });
+            if (subscribeToResult) {
+                updates.push(
+                    job.updates().pipe(
+                        map(update => {
+                            Logger.info(`Job ${update.id}: progress: ${update.progress}`);
+                            if (update.state === JobState.COMPLETED) {
+                                Logger.info(`COMPLETED: ${JSON.stringify(update.result, null, 2)}`);
+                                return update.result;
+                            }
+                            return update.progress;
+                        }),
+                        catchError(err => of(err.message)),
+                    ),
+                );
+            }
+        }
         if (subscribeToResult) {
-            return job.updates().pipe(
-                map(update => {
-                    Logger.info(`Job ${update.id}: progress: ${update.progress}`);
-                    if (update.state === JobState.COMPLETED) {
-                        Logger.info(`COMPLETED: ${JSON.stringify(update.result, null, 2)}`);
-                        return update.result;
-                    }
-                    return update.progress;
-                }),
-                catchError(err => of(err.message)),
-            );
+            return forkJoin(...updates);
         } else {
             return 'running in background';
         }
@@ -94,4 +110,9 @@ export class JobQueueTestResolver {
     },
     providers: [JobQueueTestService],
 })
-export class JobQueueTestPlugin {}
+export class JobQueueTestPlugin {
+    static init(options: { queueCount: number }) {
+        queueCount = options.queueCount;
+        return this;
+    }
+}

packages/job-queue-plugin/.gitignore

@@ -1,4 +1,4 @@
 preview/output
 yarn-error.log
-lib
+/package
 e2e/__data__/*.sqlite

packages/job-queue-plugin/README.md

@@ -1,12 +1,8 @@
 # Vendure Job Queue Plugin
 
-This plugin is under development. It will house plugins that implement alternate JobQueueStrategies aimed at handling more demanding production loads than can be comfortably handled by the database-powered DefaultJobQueuePlugin which ships with Vendure core.
+This plugin includes alternate JobQueueStrategy implementations built on different technologies.
 
 Implemented: 
 
 * The `PubSubPlugin` uses Google Cloud Pub/Sub to power the Vendure job queue. 
-
-Planned:
-
-* Redis
-* RabbitMQ
+* The `BullMQJobQueuePlugin` uses Redis via BullMQ.

packages/job-queue-plugin/build.ts

@@ -1,8 +0,0 @@
-/* tslint:disable:no-console */
-import fs from 'fs-extra';
-import path from 'path';
-
-fs.copyFileSync(
-    path.join(__dirname, './src/pub-sub/package.json'),
-    path.join(__dirname, './lib/pub-sub/package.json'),
-);

packages/job-queue-plugin/docker-compose.yml

@@ -0,0 +1,18 @@
+version: "3"
+services:
+  redis:
+    image: bitnami/redis:6.2
+    hostname: redis
+    container_name: redis
+    environment:
+      - ALLOW_EMPTY_PASSWORD=yes
+    ports:
+      - "6379:6379"
+  redis-commander:
+    container_name: redis-commander
+    hostname: redis-commander
+    image: rediscommander/redis-commander:latest
+    environment:
+      - REDIS_HOSTS=local:redis:6379
+    ports:
+      - "8085:8081"

packages/job-queue-plugin/e2e/bullmq-job-queue-plugin.e2e-spec.ts

@@ -0,0 +1,67 @@
+import { DefaultLogger, LogLevel, mergeConfig } from '@vendure/core';
+import { createTestEnvironment } from '@vendure/testing';
+import { RedisConnection } from 'bullmq';
+import path from 'path';
+
+import { initialData } from '../../../e2e-common/e2e-initial-data';
+import { testConfig, TEST_SETUP_TIMEOUT_MS } from '../../../e2e-common/test-config';
+import { awaitRunningJobs } from '../../core/e2e/utils/await-running-jobs';
+import { BullMQJobQueuePlugin } from '../src/bullmq/plugin';
+// tslint:disable-next-line:no-var-requires
+const Redis = require('ioredis');
+// tslint:disable-next-line:no-var-requires
+const { redisHost, redisPort } = require('./constants');
+
+jest.setTimeout(10 * 3000);
+
+// TODO: How to solve issues with Jest open handles after test suite finishes?
+// See https://github.com/luin/ioredis/issues/1088
+
+describe('BullMQJobQueuePlugin', () => {
+    const redisConnection: any = new Redis({
+        host: redisHost,
+        port: redisPort,
+    });
+
+    const { server, adminClient, shopClient } = createTestEnvironment(
+        mergeConfig(testConfig, {
+            apiOptions: {
+                port: 4050,
+            },
+            logger: new DefaultLogger({ level: LogLevel.Info }),
+            plugins: [
+                BullMQJobQueuePlugin.init({
+                    connection: redisConnection,
+                    workerOptions: {
+                        prefix: 'e2e',
+                    },
+                    queueOptions: {
+                        prefix: 'e2e',
+                    },
+                }),
+            ],
+        }),
+    );
+
+    beforeAll(async () => {
+        await server.init({
+            initialData,
+            productsCsvPath: path.join(__dirname, 'fixtures/e2e-products-minimal.csv'),
+            customerCount: 1,
+        });
+        await adminClient.asSuperAdmin();
+    }, TEST_SETUP_TIMEOUT_MS);
+
+    afterAll(async () => {
+        await awaitRunningJobs(adminClient);
+        await server.destroy();
+        // redis.quit() creates a thread to close the connection.
+        // We wait until all threads have been run once to ensure the connection closes.
+        // See https://stackoverflow.com/a/54560610/772859
+        await new Promise(resolve => setTimeout(resolve, 100));
+    });
+
+    it('works', () => {
+        expect(1).toBe(1);
+    });
+});

packages/job-queue-plugin/e2e/check-connection.js

@@ -0,0 +1,57 @@
+const { RedisConnection } = require('bullmq');
+const { redisHost, redisPort } = require('./constants');
+
+const connection = new RedisConnection({
+    port: redisPort,
+    host: redisHost,
+});
+
+let timer;
+
+/**
+ * When contributing to Vendure, developers who made changes unrelated to
+ * this plugin should not be expected to set up an Redis instance
+ * locally just so they can get the pre-push hook to pass. So if no
+ * instance is available, we skip the tests.
+ */
+function checkConnection() {
+    return new Promise(async (resolve, reject) => {
+        try {
+            timer = setTimeout(() => {
+                logConnectionFailure();
+                process.exit(0);
+            }, 5000);
+            connection.on('error', err => {
+                resolve(0);
+            });
+            const client = await connection.client;
+
+            clearTimeout(timer);
+
+            await client.ping((err, result) => {
+                if (err) {
+                    resolve(0);
+                } else {
+                    // If the connection is available, we exit with 1 in order to invoke the
+                    // actual e2e test script (since we are using the `||` operator in the "e2e" script)
+                    resolve(1);
+                }
+            });
+        } catch (e) {
+            logConnectionFailure();
+            // If no redis available, we exit with 0 so that the npm script
+            // exits
+            resolve(0);
+        }
+    });
+}
+
+checkConnection().then(result => {
+    process.exit(result);
+});
+
+function logConnectionFailure() {
+    console.log(`Could not connect to Redis instance at "${redisHost}:${redisPort}"`);
+    console.log(`Skipping e2e tests for BullMQJobQueuePlugin`);
+    process.env.SKIP_ELASTICSEARCH_E2E_TESTS = true;
+}

packages/job-queue-plugin/e2e/config/tsconfig.e2e.json

@@ -0,0 +1,4 @@
+{
+  "extends": "../../../../e2e-common/tsconfig.e2e.json",
+  "include": ["../**/*.e2e-spec.ts", "../**/*.d.ts"]
+}

packages/job-queue-plugin/e2e/constants.js

@@ -0,0 +1,7 @@
+const redisHost = '127.0.0.1';
+const redisPort = process.env.CI ? +(process.env.E2E_REDIS_PORT || 6379) : 6379;
+
+module.exports = {
+    redisHost,
+    redisPort,
+};

packages/job-queue-plugin/e2e/fixtures/e2e-products-minimal.csv

@@ -0,0 +1,5 @@
+name   , slug   , description                                                                                                                                                                                                                                                                  , assets                           , facets                                  , optionGroups       , optionValues    , sku      , price   , taxCategory , stockOnHand , trackInventory , variantAssets , variantFacets
+Laptop , laptop , "Now equipped with seventh-generation Intel Core processors, Laptop is snappier than ever. From daily tasks like launching apps and opening files to more advanced computing, you can power through your day thanks to faster SSDs and Turbo Boost processing up to 3.6GHz." , derick-david-409858-unsplash.jpg , category:electronics|category:computers , "screen size|RAM"  , "13 inch|8GB"   , L2201308 , 1299.00 , standard    , 100         , false          ,               ,
+       ,        ,                                                                                                                                                                                                                                                                              ,                                  ,                                         ,                    , "15 inch|8GB"   , L2201508 , 1399.00 , standard    , 100         , false          ,               ,
+       ,        ,                                                                                                                                                                                                                                                                              ,                                  ,                                         ,                    , "13 inch|16GB"  , L2201316 , 2199.00 , standard    , 100         , false          ,               ,
+       ,        ,                                                                                                                                                                                                                                                                              ,                                  ,                                         ,                    , "15 inch|16GB"  , L2201516 , 2299.00 , standard    , 100         , false          ,               ,

packages/job-queue-plugin/package.json

@@ -2,27 +2,29 @@
   "name": "@vendure/job-queue-plugin",
   "version": "1.1.3",
   "license": "MIT",
-  "main": "lib/index.js",
-  "types": "lib/index.d.ts",
+  "main": "package/index.js",
+  "types": "package/index.d.ts",
   "files": [
-    "lib/**/*"
+    "package/**/*"
   ],
-  "private": true,
+  "private": false,
   "scripts": {
     "watch": "tsc -p ./tsconfig.build.json --watch",
-    "build": "rimraf lib && tsc -p ./tsconfig.build.json && yarn ts-node build.ts",
+    "build": "rimraf package && tsc -p ./tsconfig.build.json",
     "lint": "tslint --fix --project ./",
     "test": "jest --config ./jest.config.js",
+    "e2e-wip": "node e2e/check-connection.js || jest --config ../../e2e-common/jest-config.js --runInBand --package=job-queue-plugin",
     "ci": "yarn build"
   },
   "publishConfig": {
-    "access": "restricted"
+    "access": "public"
   },
   "devDependencies": {
     "@google-cloud/pubsub": "^2.8.0",
     "@types/redis": "^2.8.28",
     "@vendure/common": "^1.1.3",
     "@vendure/core": "^1.1.3",
+    "bullmq": "^1.40.1",
     "redis": "^3.0.2",
     "rimraf": "^3.0.2",
     "typescript": "4.1.5"

packages/job-queue-plugin/src/bullmq/bullmq-job-queue-strategy.ts

@@ -0,0 +1,238 @@
+import { JobListOptions, JobState } from '@vendure/common/lib/generated-types';
+import { notNullOrUndefined } from '@vendure/common/lib/shared-utils';
+import {
+    ID,
+    Injector,
+    InspectableJobQueueStrategy,
+    InternalServerError,
+    Job,
+    JobData,
+    Logger,
+    PaginatedList,
+} from '@vendure/core';
+import Bull, { Processor, Queue, QueueScheduler, Worker } from 'bullmq';
+
+import { ALL_JOB_TYPES, BULLMQ_PLUGIN_OPTIONS, loggerCtx } from './constants';
+import { RedisHealthIndicator } from './redis-health-indicator';
+import { BullMQPluginOptions } from './types';
+
+const QUEUE_NAME = 'vendure-job-queue';
+
+/**
+ * @description
+ * This JobQueueStrategy uses [BullMQ](https://docs.bullmq.io/) to implement a push-based job queue
+ * on top of Redis. It should not be used alone, but as part of the {@link BullMQJobQueuePlugin}.
+ *
+ * @docsCategory job-queue-plugin
+ */
+export class BullMQJobQueueStrategy implements InspectableJobQueueStrategy {
+    private queue: Queue;
+    private worker: Worker;
+    private scheduler: QueueScheduler;
+    private workerProcessor: Processor;
+    private options: BullMQPluginOptions;
+    private queueNameProcessFnMap = new Map<string, (job: Job) => Promise<any>>();
+
+    async init(injector: Injector): Promise<void> {
+        const options = injector.get<BullMQPluginOptions>(BULLMQ_PLUGIN_OPTIONS);
+        this.options = options;
+
+        const redisHealthIndicator = injector.get(RedisHealthIndicator);
+        Logger.info(`Checking Redis connection...`, loggerCtx);
+        const health = await redisHealthIndicator.isHealthy('redis');
+        if (health.redis.status === 'down') {
+            Logger.error('Could not connect to Redis', loggerCtx);
+        } else {
+            Logger.info(`Connected to Redis ✔`, loggerCtx);
+        }
+
+        this.queue = new Queue(QUEUE_NAME, {
+            ...options.queueOptions,
+            connection: options.connection,
+        }).on('error', (e: any) => Logger.error(`BullMQ Queue error: ${e.message}`, loggerCtx, e.stack));
+        const client = await this.queue.client;
+
+        if (await this.queue.isPaused()) {
+            await this.queue.resume();
+        }
+
+        this.workerProcessor = async bullJob => {
+            const queueName = bullJob.name;
+            const processFn = this.queueNameProcessFnMap.get(queueName);
+            if (processFn) {
+                const job = this.createVendureJob(bullJob);
+                try {
+                    job.on('progress', _job => bullJob.updateProgress(_job.progress));
+                    const result = await processFn(job);
+                    await bullJob.updateProgress(100);
+                    return result;
+                } catch (e) {
+                    throw e;
+                }
+            }
+            throw new InternalServerError(`No processor defined for the queue "${queueName}"`);
+        };
+
+        this.scheduler = new QueueScheduler(QUEUE_NAME, {
+            ...options.schedulerOptions,
+            connection: options.connection,
+        }).on('error', (e: any) => Logger.error(`BullMQ Scheduler error: ${e.message}`, loggerCtx, e.stack));
+    }
+
+    async add<Data extends JobData<Data> = {}>(job: Job<Data>): Promise<Job<Data>> {
+        const bullJob = await this.queue.add(job.queueName, job.data, {
+            attempts: job.retries,
+        });
+        return this.createVendureJob(bullJob);
+    }
+
+    async cancelJob(jobId: string): Promise<Job | undefined> {
+        const bullJob = await this.queue.getJob(jobId);
+        if (bullJob) {
+            try {
+                await bullJob.remove();
+                return this.createVendureJob(bullJob);
+            } catch (e) {
+                Logger.error(`Error when cancelling job: ${e.message}`, loggerCtx);
+            }
+        }
+    }
+
+    async findMany(options?: JobListOptions): Promise<PaginatedList<Job>> {
+        const start = options?.skip ?? 0;
+        const end = start + (options?.take ?? 10);
+        let jobTypes = ALL_JOB_TYPES;
+        const stateFilter = options?.filter?.state;
+        if (stateFilter?.eq) {
+            switch (stateFilter.eq) {
+                case 'PENDING':
+                    jobTypes = ['wait'];
+                    break;
+                case 'RUNNING':
+                    jobTypes = ['active'];
+                    break;
+                case 'COMPLETED':
+                    jobTypes = ['completed'];
+                    break;
+                case 'RETRYING':
+                    jobTypes = ['repeat'];
+                    break;
+                case 'FAILED':
+                    jobTypes = ['failed'];
+                    break;
+                case 'CANCELLED':
+                    jobTypes = ['failed'];
+                    break;
+            }
+        }
+        const settledFilter = options?.filter?.isSettled;
+        if (settledFilter?.eq != null) {
+            jobTypes = settledFilter.eq === true ? ['completed', 'failed'] : ['wait', 'active', 'repeat'];
+        }
+        let items: Bull.Job[] = [];
+        let jobCounts: { [index: string]: number } = {};
+        try {
+            items = await this.queue.getJobs(jobTypes, start, end);
+        } catch (e) {
+            Logger.error(e.message, loggerCtx, e.stack);
+        }
+        try {
+            jobCounts = await this.queue.getJobCounts(...jobTypes);
+        } catch (e) {
+            Logger.error(e.message, loggerCtx, e.stack);
+        }
+        const totalItems = Object.values(jobCounts).reduce((sum, count) => sum + count, 0);
+
+        return Promise.resolve({
+            items: items
+                .sort((a, b) => b.timestamp - a.timestamp)
+                .map(bullJob => this.createVendureJob(bullJob)),
+            totalItems,
+        });
+    }
+
+    async findManyById(ids: ID[]): Promise<Job[]> {
+        const bullJobs = await Promise.all(ids.map(id => this.queue.getJob(id.toString())));
+        return bullJobs.filter(notNullOrUndefined).map(j => this.createVendureJob(j));
+    }
+
+    async findOne(id: ID): Promise<Job | undefined> {
+        const bullJob = await this.queue.getJob(id.toString());
+        if (bullJob) {
+            return this.createVendureJob(bullJob);
+        }
+    }
+
+    async removeSettledJobs(queueNames?: string[], olderThan?: Date): Promise<number> {
+        try {
+            const jobCounts = await this.queue.getJobCounts('completed', 'failed');
+            await this.queue.clean(100, 0, 'completed');
+            await this.queue.clean(100, 0, 'failed');
+            return Object.values(jobCounts).reduce((sum, num) => sum + num, 0);
+        } catch (e) {
+            Logger.error(e.message, loggerCtx, e.stack);
+            return 0;
+        }
+    }
+
+    async start<Data extends JobData<Data> = {}>(
+        queueName: string,
+        process: (job: Job<Data>) => Promise<any>,
+    ): Promise<void> {
+        this.queueNameProcessFnMap.set(queueName, process);
+        if (!this.worker) {
+            this.worker = new Worker(QUEUE_NAME, this.workerProcessor).on('error', (e: any) =>
+                Logger.error(`BullMQ Worker error: ${e.message}`, loggerCtx, e.stack),
+            );
+        }
+    }
+
+    async stop<Data extends JobData<Data> = {}>(
+        queueName: string,
+        process: (job: Job<Data>) => Promise<any>,
+    ): Promise<void> {
+        await this.scheduler.disconnect();
+        await this.queue.disconnect();
+        await this.worker.disconnect();
+    }
+
+    private createVendureJob(bullJob: Bull.Job): Job {
+        const jobJson = bullJob.toJSON();
+        return new Job({
+            queueName: bullJob.name,
+            id: bullJob.id,
+            state: this.getState(bullJob),
+            data: bullJob.data,
+            attempts: bullJob.attemptsMade,
+            createdAt: new Date(jobJson.timestamp),
+            startedAt: jobJson.processedOn ? new Date(jobJson.processedOn) : undefined,
+            settledAt: jobJson.finishedOn ? new Date(jobJson.finishedOn) : undefined,
+            error: jobJson.failedReason,
+            progress: +jobJson.progress,
+            result: jobJson.returnvalue,
+            retries: bullJob.opts.attempts ?? 0,
+        });
+    }
+
+    private getState(bullJob: Bull.Job): JobState {
+        const jobJson = bullJob.toJSON();
+
+        if (!jobJson.processedOn && !jobJson.failedReason) {
+            return JobState.PENDING;
+        }
+        if (!jobJson.finishedOn) {
+            return JobState.RUNNING;
+        }
+        if (jobJson.failedReason && bullJob.attemptsMade < (bullJob.opts.attempts ?? 0)) {
+            return JobState.RETRYING;
+        }
+        if (jobJson.failedReason) {
+            return JobState.FAILED;
+        }
+        if (jobJson.finishedOn) {
+            return JobState.COMPLETED;
+        }
+        throw new InternalServerError('Could not determine job state');
+        // TODO: how to handle "cancelled" state? Currently when we cancel a job, we simply remove all record of it.
+    }
+}

packages/job-queue-plugin/src/bullmq/constants.ts

@@ -0,0 +1,13 @@
+export const loggerCtx = 'BullMQJobQueuePlugin';
+export const BULLMQ_PLUGIN_OPTIONS = Symbol('BULLMQ_PLUGIN_OPTIONS');
+
+export const ALL_JOB_TYPES = [
+    'completed',
+    'failed',
+    'delayed',
+    'repeat',
+    'waiting-children',
+    'active',
+    'wait',
+    'paused',
+];

packages/job-queue-plugin/src/bullmq/index.ts

@@ -0,0 +1,14 @@
+import { Logger } from '@vendure/core';
+
+// ensure that the bullmq package is installed
+try {
+    // tslint:disable-next-line:no-var-requires
+    require('bullmq');
+} catch (e) {
+    // tslint:disable-next-line:no-console
+    console.error('The BullMQJobQueuePlugin depends on the "bullmq" package being installed.');
+    process.exit(1);
+}
+
+export * from './plugin';
+export * from './types';

packages/job-queue-plugin/src/bullmq/plugin.ts

@@ -0,0 +1,83 @@
+import { HealthCheckRegistryService, PluginCommonModule, VendurePlugin } from '@vendure/core';
+
+import { BullMQJobQueueStrategy } from './bullmq-job-queue-strategy';
+import { BULLMQ_PLUGIN_OPTIONS } from './constants';
+import { RedisHealthIndicator } from './redis-health-indicator';
+import { BullMQPluginOptions } from './types';
+
+/**
+ * @description
+ * This plugin is a drop-in replacement of the DefaultJobQueuePlugin, which implements a push-based
+ * job queue strategy built on top of the popular [BullMQ](https://github.com/taskforcesh/bullmq) library.
+ *
+ * {{% alert "warning" %}}
+ * This plugin was newly released with Vendure v1.2.0 and has yet to receive thorough real-world testing.
+ * {{% /alert %}}
+ *
+ * ## Advantages over the DefaultJobQueuePlugin
+ *
+ * The advantage of this approach is that jobs are stored in Redis rather than in the database. For more complex
+ * applications with many job queues and/or multiple worker instances, this can massively reduce the load on the
+ * DB server. The reason for this is that the DefaultJobQueuePlugin uses polling to check for new jobs. By default
+ * it will poll every 200ms. A typical Vendure instance uses at least 3 queues (handling emails, collections, search index),
+ * so even with a single worker instance this results in 15 queries per second to the DB constantly. Adding more
+ * custom queues and multiple worker instances can easily result in 50 or 100 queries per second. At this point
+ * performance may be impacted.
+ *
+ * Using this plugin, no polling is needed, as BullMQ will _push_ jobs to the worker(s) as and when they are added
+ * to the queue. This results in significantly more scalable performance characteristics, as well as lower latency
+ * in processing jobs.
+ *
+ * ## Installation
+ *
+ * `yarn add \@vendure/job-queue-plugin bullmq`
+ *
+ * or
+ *
+ * `npm install \@vendure/job-queue-plugin bullmq`
+ *
+ * @example
+ * ```ts
+ * import { BullMQJobQueuePlugin } from '\@vendure/job-queue-plugin/packages/bullmq';
+ *
+ * const config: VendureConfig = {
+ *   // Add an instance of the plugin to the plugins array
+ *   plugins: [
+ *     BullMQJobQueuePlugin.init({
+ *       connection: {
+ *         port: 6379
+ *       }
+ *     }),
+ *   ],
+ * };
+ * ```
+ *
+ * @docsCategory job-queue-plugin
+ */
+@VendurePlugin({
+    imports: [PluginCommonModule],
+    configuration: config => {
+        config.jobQueueOptions.jobQueueStrategy = new BullMQJobQueueStrategy();
+        return config;
+    },
+    providers: [
+        { provide: BULLMQ_PLUGIN_OPTIONS, useFactory: () => BullMQJobQueuePlugin.options },
+        RedisHealthIndicator,
+    ],
+})
+export class BullMQJobQueuePlugin {
+    static options: BullMQPluginOptions;
+
+    /**
+     * @description
+     * Configures the plugin.
+     */
+    static init(options: BullMQPluginOptions) {
+        this.options = options;
+        return this;
+    }
+
+    constructor(private registry: HealthCheckRegistryService, private redis: RedisHealthIndicator) {
+        registry.registerIndicatorFunction(() => this.redis.isHealthy('redis (job queue)'));
+    }
+}

packages/job-queue-plugin/src/bullmq/redis-health-indicator.ts

@@ -0,0 +1,66 @@
+import { Inject, Injectable } from '@nestjs/common';
+import { HealthCheckError, HealthIndicator, HealthIndicatorResult } from '@nestjs/terminus';
+import { Logger } from '@vendure/core';
+import { RedisConnection } from 'bullmq';
+import { timer } from 'rxjs';
+
+import { BULLMQ_PLUGIN_OPTIONS, loggerCtx } from './constants';
+import { BullMQPluginOptions } from './types';
+
+@Injectable()
+export class RedisHealthIndicator extends HealthIndicator {
+    private timeoutTimer: any;
+    constructor(@Inject(BULLMQ_PLUGIN_OPTIONS) private options: BullMQPluginOptions) {
+        super();
+    }
+    async isHealthy(key: string, timeoutMs = 5000): Promise<HealthIndicatorResult> {
+        let connection: RedisConnection;
+        connection = new RedisConnection({
+            ...this.options.connection,
+            connectTimeout: 10000,
+        });
+        const pingResult = await new Promise(async (resolve, reject) => {
+            try {
+                connection.on('error', err => {
+                    Logger.error(`Redis health check error: ${err.message}`, loggerCtx, err.stack);
+                    resolve(err);
+                });
+                if (this.timeoutTimer) {
+                    clearTimeout(this.timeoutTimer);
+                }
+                const timeout = new Promise<void>(
+                    _resolve => (this.timeoutTimer = setTimeout(_resolve, timeoutMs)),
+                );
+                const client = await Promise.race([connection.client, timeout]);
+                clearTimeout(this.timeoutTimer);
+                if (!client) {
+                    resolve('timeout');
+                    return;
+                }
+                client.ping((err, res) => {
+                    if (err) {
+                        resolve(err);
+                    } else {
+                        resolve(res);
+                    }
+                });
+            } catch (e) {
+                resolve(e);
+            }
+        });
+
+        try {
+            await connection.close();
+            // await connection.disconnect();
+        } catch (e) {
+            Logger.error(`Redis health check error closing connection: ${e.message}`, loggerCtx, e.stack);
+        }
+
+        const result = this.getStatus(key, pingResult === 'PONG');
+
+        if (pingResult) {
+            return result;
+        }
+        throw new HealthCheckError('Redis failed', result);
+    }
+}

packages/job-queue-plugin/src/bullmq/types.ts

@@ -0,0 +1,38 @@
+import { ConnectionOptions, QueueSchedulerOptions, WorkerOptions } from 'bullmq';
+import { QueueOptions } from 'bullmq';
+
+/**
+ * @description
+ * Configuration options for the {@link BullMQJobQueuePlugin}.
+ *
+ * @since 1.2.0
+ * @docsCategory job-queue-plugin
+ */
+export interface BullMQPluginOptions {
+    /**
+     * @description
+     * Connection options which will be passed directly to BullMQ when
+     * creating a new Queue, Worker and Scheduler instance.
+     *
+     * If omitted, it will attempt to connect to Redis at `127.0.0.1:6379`.
+     */
+    connection?: ConnectionOptions;
+    /**
+     * @description
+     * Additional options used when instantiating the BullMQ
+     * Queue instance.
+     */
+    queueOptions?: Exclude<QueueOptions, 'connection'>;
+    /**
+     * @description
+     * Additional options used when instantiating the BullMQ
+     * Worker instance.
+     */
+    workerOptions?: Exclude<WorkerOptions, 'connection'>;
+    /**
+     * @description
+     * Additional options used when instantiating the BullMQ
+     * QueueScheduler instance.
+     */
+    schedulerOptions?: Exclude<QueueSchedulerOptions, 'connection'>;
+}

packages/job-queue-plugin/src/pub-sub/package.json

@@ -1,5 +0,0 @@
-{
-  "main": "index.js",
-  "typings": "index.d.ts",
-  "name": "@vendure/job-queue-plugin/pub-sub"
-}

packages/job-queue-plugin/tsconfig.build.json

@@ -1,7 +1,7 @@
 {
   "extends": "./tsconfig.json",
   "compilerOptions": {
-    "outDir": "./lib"
+    "outDir": "./package"
   },
   "include": [
     "./src/**/index.ts"