Merge branch 'master' into minor

docs/content/developer-guide/importing-product-data.md

@@ -128,7 +128,7 @@ The `@vendure/core` package exposes a [`populate()` function]({{< relref "popula
 
 ```TypeScript
 // populate-server.ts
-import { bootstrap } from '@vendure/core';
+import { bootstrap, DefaultJobQueuePlugin } from '@vendure/core';
 import { populate } from '@vendure/core/cli';
 
 import { config } from './vendure-config.ts';
@@ -136,6 +136,15 @@ import { initialData } from './my-initial-data.ts';
 
 const productsCsvFile = path.join(__dirname, 'path/to/products.csv')
 
+const populateConfig = {
+  ...config,
+  plugins: (config.plugins || []).filter(
+    // Remove your JobQueuePlugin during populating to avoid
+    // generating lots of unnecessary jobs as the Collections get created.
+    plugin => plugin !== DefaultJobQueuePlugin,
+  ),
+}
+
 populate(
   () => bootstrap(config),
   initialData,

packages/admin-ui/src/lib/core/src/common/generated-types.ts

@@ -5541,7 +5541,7 @@ export type GetCustomerQuery = { customer?: Maybe<(
       & Pick<OrderList, 'totalItems'>
       & { items: Array<(
         { __typename?: 'Order' }
-        & Pick<Order, 'id' | 'code' | 'state' | 'total' | 'currencyCode' | 'updatedAt'>
+        & Pick<Order, 'id' | 'code' | 'state' | 'totalWithTax' | 'currencyCode' | 'updatedAt'>
       )> }
     ) }
     & CustomerFragment

packages/admin-ui/src/lib/core/src/data/definitions/customer-definitions.ts

@@ -82,7 +82,7 @@ export const GET_CUSTOMER = gql`
                     id
                     code
                     state
-                    total
+                    totalWithTax
                     currencyCode
                     updatedAt
                 }

packages/admin-ui/src/lib/core/src/data/providers/administrator-data.service.ts

@@ -51,12 +51,8 @@ export class AdministratorDataService {
         );
     }
 
-    getActiveAdministrator(fetchPolicy: FetchPolicy = 'cache-first') {
-        return this.baseDataService.query<GetActiveAdministrator.Query>(
-            GET_ACTIVE_ADMINISTRATOR,
-            {},
-            fetchPolicy,
-        );
+    getActiveAdministrator() {
+        return this.baseDataService.query<GetActiveAdministrator.Query>(GET_ACTIVE_ADMINISTRATOR, {});
     }
 
     getAdministrator(id: string) {

packages/admin-ui/src/lib/customer/src/components/customer-detail/customer-detail.component.html

@@ -141,7 +141,7 @@
             <ng-template let-order="item">
                 <td class="left">{{ order.code }}</td>
                 <td class="left">{{ order.state }}</td>
-                <td class="left">{{ order.total | localeCurrency: order.currencyCode }}</td>
+                <td class="left">{{ order.totalWithTax | localeCurrency: order.currencyCode }}</td>
                 <td class="left">{{ order.updatedAt | localeDate: 'medium' }}</td>
                 <td class="right">
                     <vdr-table-row-action

packages/core/e2e/graphql/fragments.ts

@@ -35,6 +35,8 @@ export const ASSET_FRAGMENT = gql`
 export const PRODUCT_VARIANT_FRAGMENT = gql`
     fragment ProductVariant on ProductVariant {
         id
+        createdAt
+        updatedAt
         enabled
         languageCode
         name

packages/core/e2e/graphql/generated-e2e-admin-types.ts

@@ -5253,6 +5253,8 @@ export type AssetFragment = Pick<
 export type ProductVariantFragment = Pick<
     ProductVariant,
     | 'id'
+    | 'createdAt'
+    | 'updatedAt'
     | 'enabled'
     | 'languageCode'
     | 'name'

packages/core/e2e/product.e2e-spec.ts

@@ -1371,6 +1371,33 @@ describe('Product resolver', () => {
                 expect(updatedVariant.price).toBe(432);
             });
 
+            // https://github.com/vendure-ecommerce/vendure/issues/1101
+            it('after update, the updatedAt should be modified', async () => {
+                // Pause for a second to ensure the updatedAt date is more than 1s
+                // later than the createdAt date, since sqlite does not seem to store
+                // down to millisecond resolution.
+                await new Promise(resolve => setTimeout(resolve, 1000));
+
+                const firstVariant = variants[0];
+                const { updateProductVariants } = await adminClient.query<
+                    UpdateProductVariants.Mutation,
+                    UpdateProductVariants.Variables
+                >(UPDATE_PRODUCT_VARIANTS, {
+                    input: [
+                        {
+                            id: firstVariant.id,
+                            translations: firstVariant.translations,
+                            sku: 'ABCD',
+                            price: 432,
+                        },
+                    ],
+                });
+
+                const updatedVariant = updateProductVariants.find(v => v?.id === variants[0].id);
+
+                expect(updatedVariant?.updatedAt).not.toBe(updatedVariant?.createdAt);
+            });
+
             it('updateProductVariants updates assets', async () => {
                 const firstVariant = variants[0];
                 const result = await adminClient.query<
@@ -1590,7 +1617,7 @@ describe('Product resolver', () => {
 
     describe('deletion', () => {
         let allProducts: GetProductList.Items[];
-        let productToDelete: GetProductList.Items;
+        let productToDelete: GetProductWithVariants.Product;
 
         beforeAll(async () => {
             const result = await adminClient.query<GetProductList.Query, GetProductList.Variables>(
@@ -1607,24 +1634,45 @@ describe('Product resolver', () => {
         });
 
         it('deletes a product', async () => {
-            productToDelete = allProducts[0];
+            const { product } = await adminClient.query<
+                GetProductWithVariants.Query,
+                GetProductWithVariants.Variables
+            >(GET_PRODUCT_WITH_VARIANTS, {
+                id: allProducts[0].id,
+            });
             const result = await adminClient.query<DeleteProduct.Mutation, DeleteProduct.Variables>(
                 DELETE_PRODUCT,
-                { id: productToDelete.id },
+                { id: product!.id },
             );
 
             expect(result.deleteProduct).toEqual({ result: DeletionResult.DELETED });
+
+            productToDelete = product!;
         });
 
         it('cannot get a deleted product', async () => {
-            const result = await adminClient.query<
+            const { product } = await adminClient.query<
                 GetProductWithVariants.Query,
                 GetProductWithVariants.Variables
             >(GET_PRODUCT_WITH_VARIANTS, {
                 id: productToDelete.id,
             });
 
-            expect(result.product).toBe(null);
+            expect(product).toBe(null);
+        });
+
+        // https://github.com/vendure-ecommerce/vendure/issues/1096
+        it('variants of deleted product are also deleted', async () => {
+            for (const variant of productToDelete.variants) {
+                const { productVariant } = await adminClient.query<
+                    GetProductVariant.Query,
+                    GetProductVariant.Variables
+                >(GET_PRODUCT_VARIANT, {
+                    id: variant.id,
+                });
+
+                expect(productVariant).toBe(null);
+            }
         });
 
         it('deleted product omitted from list', async () => {

packages/core/src/data-import/providers/populator/populator.ts

@@ -4,7 +4,7 @@ import { normalizeString } from '@vendure/common/lib/normalize-string';
 import { notNullOrUndefined } from '@vendure/common/lib/shared-utils';
 
 import { RequestContext } from '../../../api/common/request-context';
-import { defaultShippingCalculator, defaultShippingEligibilityChecker } from '../../../config';
+import { defaultShippingCalculator, defaultShippingEligibilityChecker, Logger } from '../../../config';
 import { manualFulfillmentHandler } from '../../../config/fulfillment/manual-fulfillment-handler';
 import { Channel, Collection, FacetValue, TaxCategory } from '../../../entity';
 import {
@@ -55,13 +55,44 @@ export class Populator {
      */
     async populateInitialData(data: InitialData) {
         const { channel, ctx } = await this.createRequestContext(data);
-
-        const zoneMap = await this.populateCountries(ctx, data.countries);
-        await this.populateTaxRates(ctx, data.taxRates, zoneMap);
-        await this.populateShippingMethods(ctx, data.shippingMethods);
-        await this.populatePaymentMethods(ctx, data.paymentMethods);
-        await this.setChannelDefaults(zoneMap, data, channel);
-        await this.populateRoles(ctx, data.roles);
+        let zoneMap: ZoneMap;
+        try {
+            zoneMap = await this.populateCountries(ctx, data.countries);
+        } catch (e) {
+            Logger.error(`Could not populate countries`);
+            Logger.error(e, 'populator', e.stack);
+            throw e;
+        }
+        try {
+            await this.populateTaxRates(ctx, data.taxRates, zoneMap);
+        } catch (e) {
+            Logger.error(`Could not populate tax rates`);
+            Logger.error(e, 'populator', e.stack);
+        }
+        try {
+            await this.populateShippingMethods(ctx, data.shippingMethods);
+        } catch (e) {
+            Logger.error(`Could not populate shipping methods`);
+            Logger.error(e, 'populator', e.stack);
+        }
+        try {
+            await this.populatePaymentMethods(ctx, data.paymentMethods);
+        } catch (e) {
+            Logger.error(`Could not populate payment methods`);
+            Logger.error(e, 'populator', e.stack);
+        }
+        try {
+            await this.setChannelDefaults(zoneMap, data, channel);
+        } catch (e) {
+            Logger.error(`Could not set channel defaults`);
+            Logger.error(e, 'populator', e.stack);
+        }
+        try {
+            await this.populateRoles(ctx, data.roles);
+        } catch (e) {
+            Logger.error(`Could not populate roles`);
+            Logger.error(e, 'populator', e.stack);
+        }
     }
 
     /**

packages/core/src/job-queue/job.ts

@@ -1,6 +1,8 @@
 import { JobState } from '@vendure/common/lib/generated-types';
 import { isClassInstance, isObject } from '@vendure/common/lib/shared-utils';
 
+import { Logger } from '../config/logger/vendure-logger';
+
 import { JobConfig, JobData } from './types';
 
 /**
@@ -124,6 +126,11 @@ export class Job<T extends JobData<T> = any> {
             this._state = JobState.RUNNING;
             this._startedAt = new Date();
             this._attempts++;
+            Logger.debug(
+                `Job ${this.id} [${this.queueName}] starting (attempt ${this._attempts} of ${
+                    this.retries + 1
+                })`,
+            );
         }
     }
 
@@ -146,6 +153,7 @@ export class Job<T extends JobData<T> = any> {
         this._progress = 100;
         this._state = JobState.COMPLETED;
         this._settledAt = new Date();
+        Logger.debug(`Job ${this.id} [${this.queueName}] completed`);
     }
 
     /**
@@ -157,9 +165,15 @@ export class Job<T extends JobData<T> = any> {
         this._progress = 0;
         if (this.retries >= this._attempts) {
             this._state = JobState.RETRYING;
+            Logger.warn(
+                `Job ${this.id} [${this.queueName}] failed (attempt ${this._attempts} of ${
+                    this.retries + 1
+                })`,
+            );
         } else {
             if (this._state !== JobState.CANCELLED) {
                 this._state = JobState.FAILED;
+                Logger.warn(`Job ${this.id} [${this.queueName}] failed and will not retry.`);
             }
             this._settledAt = new Date();
         }
@@ -179,6 +193,7 @@ export class Job<T extends JobData<T> = any> {
         if (this._state === JobState.RUNNING) {
             this._state = JobState.PENDING;
             this._attempts = 0;
+            Logger.debug(`Job ${this.id} [${this.queueName}] deferred back to PENDING state`);
         }
     }
 

packages/core/src/job-queue/polling-job-queue-strategy.ts

@@ -23,8 +23,26 @@ import { JobData } from './types';
 export type BackoffStrategy = (queueName: string, attemptsMade: number, job: Job) => number;
 
 export interface PollingJobQueueStrategyConfig {
+    /**
+     * @description
+     * How many jobs from a given queue to process concurrently.
+     *
+     * @default 1
+     */
     concurrency?: number;
+    /**
+     * @description
+     * The interval in ms between polling the database for new jobs.
+     *
+     * @description 200
+     */
     pollInterval?: number | ((queueName: string) => number);
+    /**
+     * @description
+     * The strategy used to decide how long to wait before retrying a failed job.
+     *
+     * @default () => 1000
+     */
     backoffStrategy?: BackoffStrategy;
 }
 
@@ -177,7 +195,7 @@ export abstract class PollingJobQueueStrategy extends InjectableJobQueueStrategy
         if (concurrencyOrConfig && isObject(concurrencyOrConfig)) {
             this.concurrency = concurrencyOrConfig.concurrency ?? 1;
             this.pollInterval = concurrencyOrConfig.pollInterval ?? 200;
-            this.backOffStrategy = concurrencyOrConfig.backoffStrategy;
+            this.backOffStrategy = concurrencyOrConfig.backoffStrategy ?? (() => 1000);
         } else {
             this.concurrency = concurrencyOrConfig ?? 1;
             this.pollInterval = maybePollInterval ?? 200;

packages/core/src/service/helpers/utils/translate-entity.ts

@@ -69,7 +69,7 @@ export function translateEntity<T extends Translatable & VendureEntity>(
                 translated.customFields = {};
             }
             Object.assign(translated.customFields, value);
-        } else if (key !== 'base' && key !== 'id') {
+        } else if (key !== 'base' && key !== 'id' && key !== 'createdAt' && key !== 'updatedAt') {
             translated[key] = value;
         }
     }

packages/core/src/service/services/customer.service.ts

@@ -251,33 +251,36 @@ export class CustomerService {
         });
 
         if (hasEmailAddress(input)) {
-            const existingCustomerInChannel = await this.connection
-                .getRepository(ctx, Customer)
-                .createQueryBuilder('customer')
-                .leftJoin('customer.channels', 'channel')
-                .where('channel.id = :channelId', { channelId: ctx.channelId })
-                .andWhere('customer.emailAddress = :emailAddress', { emailAddress: input.emailAddress })
-                .andWhere('customer.id != :customerId', { customerId: input.id })
-                .andWhere('customer.deletedAt is null')
-                .getOne();
-
-            if (existingCustomerInChannel) {
-                return new EmailAddressConflictAdminError();
-            }
+            if (input.emailAddress !== customer.emailAddress) {
+                const existingCustomerInChannel = await this.connection
+                    .getRepository(ctx, Customer)
+                    .createQueryBuilder('customer')
+                    .leftJoin('customer.channels', 'channel')
+                    .where('channel.id = :channelId', { channelId: ctx.channelId })
+                    .andWhere('customer.emailAddress = :emailAddress', { emailAddress: input.emailAddress })
+                    .andWhere('customer.id != :customerId', { customerId: input.id })
+                    .andWhere('customer.deletedAt is null')
+                    .getOne();
+
+                if (existingCustomerInChannel) {
+                    return new EmailAddressConflictAdminError();
+                }
 
-            if (customer.user) {
-                const existingUserWithEmailAddress = await this.userService.getUserByEmailAddress(
-                    ctx,
-                    input.emailAddress,
-                );
+                if (customer.user) {
+                    const existingUserWithEmailAddress = await this.userService.getUserByEmailAddress(
+                        ctx,
+                        input.emailAddress,
+                    );
 
-                if (
-                    existingUserWithEmailAddress &&
-                    !idsAreEqual(customer.user.id, existingUserWithEmailAddress.id)
-                ) {
-                    return new EmailAddressConflictAdminError();
+                    if (
+                        existingUserWithEmailAddress &&
+                        !idsAreEqual(customer.user.id, existingUserWithEmailAddress.id)
+                    ) {
+                        return new EmailAddressConflictAdminError();
+                    }
+
+                    await this.userService.changeNativeIdentifier(ctx, customer.user.id, input.emailAddress);
                 }
-                await this.userService.changeNativeIdentifier(ctx, customer.user.id, input.emailAddress);
             }
         }
 

packages/core/src/service/services/product-variant.service.ts

@@ -102,7 +102,10 @@ export class ProductVariantService {
     findOne(ctx: RequestContext, productVariantId: ID): Promise<Translated<ProductVariant> | undefined> {
         const relations = ['product', 'product.featuredAsset', 'taxCategory'];
         return this.connection
-            .findOneInChannel(ctx, ProductVariant, productVariantId, ctx.channelId, { relations })
+            .findOneInChannel(ctx, ProductVariant, productVariantId, ctx.channelId, {
+                relations,
+                where: { deletedAt: null },
+            })
             .then(async result => {
                 if (result) {
                     return translateDeep(await this.applyChannelPriceAndTax(result, ctx), ctx.languageCode, [
@@ -321,7 +324,7 @@ export class ProductVariantService {
             ids.push(id);
         }
         const createdVariants = await this.findByIds(ctx, ids);
-        this.eventBus.publish(new ProductVariantEvent(ctx, createdVariants, 'updated'));
+        this.eventBus.publish(new ProductVariantEvent(ctx, createdVariants, 'created'));
         return createdVariants;
     }
 
@@ -498,11 +501,14 @@ export class ProductVariantService {
         return this.connection.getRepository(ctx, ProductVariantPrice).save(variantPrice);
     }
 
-    async softDelete(ctx: RequestContext, id: ID): Promise<DeletionResponse> {
-        const variant = await this.connection.getEntityOrThrow(ctx, ProductVariant, id);
-        variant.deletedAt = new Date();
-        await this.connection.getRepository(ctx, ProductVariant).save(variant, { reload: false });
-        this.eventBus.publish(new ProductVariantEvent(ctx, [variant], 'deleted'));
+    async softDelete(ctx: RequestContext, id: ID | ID[]): Promise<DeletionResponse> {
+        const ids = Array.isArray(id) ? id : [id];
+        const variants = await this.connection.getRepository(ctx, ProductVariant).findByIds(ids);
+        for (const variant of variants) {
+            variant.deletedAt = new Date();
+        }
+        await this.connection.getRepository(ctx, ProductVariant).save(variants, { reload: false });
+        this.eventBus.publish(new ProductVariantEvent(ctx, variants, 'deleted'));
         return {
             result: DeletionResult.DELETED,
         };

packages/core/src/service/services/product.service.ts

@@ -220,10 +220,15 @@ export class ProductService {
     async softDelete(ctx: RequestContext, productId: ID): Promise<DeletionResponse> {
         const product = await this.connection.getEntityOrThrow(ctx, Product, productId, {
             channelId: ctx.channelId,
+            relations: ['variants'],
         });
         product.deletedAt = new Date();
         await this.connection.getRepository(ctx, Product).save(product, { reload: false });
         this.eventBus.publish(new ProductEvent(ctx, product, 'deleted'));
+        await this.productVariantService.softDelete(
+            ctx,
+            product.variants.map(v => v.id),
+        );
         return {
             result: DeletionResult.DELETED,
         };

packages/core/src/service/services/user.service.ts

@@ -210,7 +210,14 @@ export class UserService {
         if (!user) {
             return;
         }
-        const nativeAuthMethod = user.getNativeAuthenticationMethod();
+        const nativeAuthMethod = user.authenticationMethods.find(
+            (m): m is NativeAuthenticationMethod => m instanceof NativeAuthenticationMethod,
+        );
+        if (!nativeAuthMethod) {
+            // If the NativeAuthenticationMethod is not configured, then
+            // there is nothing to do.
+            return;
+        }
         user.identifier = newIdentifier;
         nativeAuthMethod.identifier = newIdentifier;
         nativeAuthMethod.identifierChangeToken = null;

packages/elasticsearch-plugin/src/elasticsearch.service.ts

@@ -19,7 +19,7 @@ import equal from 'fast-deep-equal/es6';
 import { buildElasticBody } from './build-elastic-body';
 import { ELASTIC_SEARCH_OPTIONS, loggerCtx, VARIANT_INDEX_NAME } from './constants';
 import { ElasticsearchIndexService } from './elasticsearch-index.service';
-import { createIndices } from './indexing-utils';
+import { createIndices, getClient } from './indexing-utils';
 import { ElasticsearchOptions } from './options';
 import {
     CustomMapping,
@@ -48,14 +48,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
     }
 
     onModuleInit(): any {
-        const { host, port } = this.options;
-        const node = this.options.clientOptions?.node ?? `${host}:${port}`;
-        this.client = new Client({
-            node,
-            // `any` cast is there due to a strange error "Property '[Symbol.iterator]' is missing in type... URLSearchParams"
-            // which looks like possibly a TS/definitions bug.
-            ...(this.options.clientOptions as any),
-        });
+        this.client = getClient(this.options);
     }
 
     onModuleDestroy(): any {
@@ -267,7 +260,13 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
         enabledOnly: boolean = false,
     ): Promise<Array<{ facetValue: FacetValue; count: number }>> {
         const { groupByProduct } = input;
-        const buckets = await this.getDistinctBucketsOfField(ctx, input, enabledOnly, `facetValueIds`,this.options.searchConfig.facetValueMaxSize);
+        const buckets = await this.getDistinctBucketsOfField(
+            ctx,
+            input,
+            enabledOnly,
+            `facetValueIds`,
+            this.options.searchConfig.facetValueMaxSize,
+        );
 
         const facetValues = await this.facetValueService.findByIds(
             ctx,
@@ -297,7 +296,13 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
         enabledOnly: boolean = false,
     ): Promise<Array<{ collection: Collection; count: number }>> {
         const { groupByProduct } = input;
-        const buckets = await this.getDistinctBucketsOfField(ctx, input, enabledOnly, `collectionIds`,this.options.searchConfig.collectionMaxSize);
+        const buckets = await this.getDistinctBucketsOfField(
+            ctx,
+            input,
+            enabledOnly,
+            `collectionIds`,
+            this.options.searchConfig.collectionMaxSize,
+        );
 
         const collections = await this.collectionService.findByIds(
             ctx,
@@ -340,7 +345,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
             aggregation_field: {
                 terms: {
                     field,
-                    size: aggregation_max_size
+                    size: aggregation_max_size,
                 },
             },
         };

packages/elasticsearch-plugin/src/indexer.controller.ts

@@ -22,8 +22,8 @@ import {
 } from '@vendure/core';
 import { Observable } from 'rxjs';
 
-import { ELASTIC_SEARCH_OPTIONS, loggerCtx, VARIANT_INDEX_NAME } from './constants';
-import { createIndices, getIndexNameByAlias } from './indexing-utils';
+import { ELASTIC_SEARCH_OPTIONS, loggerCtx, PRODUCT_INDEX_NAME, VARIANT_INDEX_NAME } from './constants';
+import { createIndices, getClient, getIndexNameByAlias } from './indexing-utils';
 import { ElasticsearchOptions } from './options';
 import {
     BulkOperation,
@@ -83,10 +83,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
     ) {}
 
     onModuleInit(): any {
-        const { host, port } = this.options;
-        this.client = new Client({
-            node: `${host}:${port}`,
-        });
+        this.client = getClient(this.options);
     }
 
     onModuleDestroy(): any {

packages/elasticsearch-plugin/src/indexing-utils.ts

@@ -1,7 +1,8 @@
 import { Client } from '@elastic/elasticsearch';
-import { ID, Logger } from '@vendure/core';
+import { DeepRequired, ID, Logger } from '@vendure/core';
 
 import { loggerCtx, VARIANT_INDEX_NAME } from './constants';
+import { ElasticsearchOptions } from './options';
 import { VariantIndexItem } from './types';
 
 export async function createIndices(
@@ -131,6 +132,19 @@ export async function deleteByChannel(client: Client, prefix: string, channelId:
     }
 }
 
+export function getClient(
+    options: Required<ElasticsearchOptions> | DeepRequired<ElasticsearchOptions>,
+): Client {
+    const { host, port } = options;
+    const node = options.clientOptions?.node ?? `${host}:${port}`;
+    return new Client({
+        node,
+        // `any` cast is there due to a strange error "Property '[Symbol.iterator]' is missing in type... URLSearchParams"
+        // which looks like possibly a TS/definitions bug.
+        ...(options.clientOptions as any),
+    });
+}
+
 export async function getIndexNameByAlias(client: Client, aliasName: string) {
     const aliasExist = await client.indices.existsAlias({ name: aliasName });
     if (aliasExist.body) {

packages/email-plugin/src/email-processor.ts

@@ -87,7 +87,7 @@ export class EmailProcessor {
             } else {
                 Logger.error(String(err), loggerCtx);
             }
-            return false;
+            throw err;
         }
     }
 }

packages/email-plugin/src/plugin.ts

@@ -247,7 +247,7 @@ export class EmailPlugin implements OnApplicationBootstrap, NestModule {
                 return;
             }
             if (this.jobQueue) {
-                await this.jobQueue.add(result);
+                await this.jobQueue.add(result, { retries: 5 });
             } else if (this.testingProcessor) {
                 await this.testingProcessor.process(result);
             }

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

@@ -51,7 +51,6 @@ export class BullMQJobQueueStrategy implements InspectableJobQueueStrategy {
             ...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();
@@ -59,6 +58,11 @@ export class BullMQJobQueueStrategy implements InspectableJobQueueStrategy {
 
         this.workerProcessor = async bullJob => {
             const queueName = bullJob.name;
+            Logger.debug(
+                `Job ${bullJob.id} [${queueName}] starting (attempt ${bullJob.attemptsMade + 1} of ${
+                    bullJob.opts.attempts ?? 1
+                })`,
+            );
             const processFn = this.queueNameProcessFnMap.get(queueName);
             if (processFn) {
                 const job = this.createVendureJob(bullJob);
@@ -82,7 +86,11 @@ export class BullMQJobQueueStrategy implements InspectableJobQueueStrategy {
 
     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,
+            attempts: job.retries + 1,
+            backoff: {
+                delay: 1000,
+                type: 'exponential',
+            },
         });
         return this.createVendureJob(bullJob);
     }
@@ -185,10 +193,22 @@ export class BullMQJobQueueStrategy implements InspectableJobQueueStrategy {
             const options: WorkerOptions = {
                 concurrency: DEFAULT_CONCURRENCY,
                 ...this.options.workerOptions,
+                connection: this.options.connection,
             };
-            this.worker = new Worker(QUEUE_NAME, this.workerProcessor, options).on('error', (e: any) =>
-                Logger.error(`BullMQ Worker error: ${e.message}`, loggerCtx, e.stack),
-            );
+            this.worker = new Worker(QUEUE_NAME, this.workerProcessor, options)
+                .on('error', (e: any) =>
+                    Logger.error(`BullMQ Worker error: ${e.message}`, loggerCtx, e.stack),
+                )
+                .on('failed', (job: Bull.Job, failedReason: string) => {
+                    Logger.warn(
+                        `Job ${job.id} [${job.name}] failed (attempt ${job.attemptsMade} of ${
+                            job.opts.attempts ?? 1
+                        })`,
+                    );
+                })
+                .on('completed', (job: Bull.Job, failedReason: string) => {
+                    Logger.debug(`Job ${job.id} [${job.name}] completed`);
+                });
         }
     }
 

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

@@ -14,11 +14,7 @@ export class RedisHealthIndicator extends HealthIndicator {
         super();
     }
     async isHealthy(key: string, timeoutMs = 5000): Promise<HealthIndicatorResult> {
-        let connection: RedisConnection;
-        connection = new RedisConnection({
-            ...this.options.connection,
-            connectTimeout: 10000,
-        });
+        const connection = new RedisConnection(this.options.connection);
         const pingResult = await new Promise(async (resolve, reject) => {
             try {
                 connection.on('error', err => {