refactor(elasticsearch-plugin) Remove use of Worker from elastic plugin

packages/elasticsearch-plugin/e2e/elasticsearch-plugin-uuid.e2e-spec.ts

@@ -29,11 +29,6 @@ describe('Elasticsearch plugin with UuidIdStrategy', () => {
             apiOptions: {
                 port: 4050,
             },
-            workerOptions: {
-                options: {
-                    port: 4055,
-                },
-            },
             entityIdStrategy: new UuidIdStrategy(),
             logger: new DefaultLogger({ level: LogLevel.Info }),
             plugins: [

packages/elasticsearch-plugin/e2e/elasticsearch-plugin.e2e-spec.ts

@@ -102,11 +102,6 @@ describe('Elasticsearch plugin', () => {
             apiOptions: {
                 port: 4050,
             },
-            workerOptions: {
-                options: {
-                    port: 4055,
-                },
-            },
             logger: new DefaultLogger({ level: LogLevel.Info }),
             plugins: [
                 ElasticsearchPlugin.init({

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

@@ -1,4 +1,4 @@
-import { Injectable } from '@nestjs/common';
+import { Injectable, OnApplicationBootstrap } from '@nestjs/common';
 import { assertNever } from '@vendure/common/lib/shared-utils';
 import {
     Asset,
@@ -10,64 +10,52 @@ import {
     Product,
     ProductVariant,
     RequestContext,
-    WorkerMessage,
-    WorkerService,
 } from '@vendure/core';
+import { Observable } from 'rxjs';
 
-import { ReindexMessageResponse } from './indexer.controller';
-import {
-    AssignProductToChannelMessage,
-    AssignVariantToChannelMessage,
-    DeleteAssetMessage,
-    DeleteProductMessage,
-    DeleteVariantMessage,
-    ReindexMessage,
-    RemoveProductFromChannelMessage,
-    RemoveVariantFromChannelMessage,
-    UpdateAssetMessage,
-    UpdateIndexQueueJobData,
-    UpdateProductMessage,
-    UpdateVariantMessage,
-    UpdateVariantsByIdMessage,
-} from './types';
-
-let updateIndexQueue: JobQueue<UpdateIndexQueueJobData> | undefined;
+import { ElasticsearchIndexerController, ReindexMessageResponse } from './indexer.controller';
+import { UpdateIndexQueueJobData } from './types';
 
 @Injectable()
-export class ElasticsearchIndexService {
-    constructor(private workerService: WorkerService, private jobService: JobQueueService) {}
+export class ElasticsearchIndexService implements OnApplicationBootstrap {
+    private updateIndexQueue: JobQueue<UpdateIndexQueueJobData>;
+
+    constructor(
+        private jobService: JobQueueService,
+        private indexerController: ElasticsearchIndexerController,
+    ) {}
 
-    initJobQueue() {
-        updateIndexQueue = this.jobService.createQueue({
+    onApplicationBootstrap() {
+        this.updateIndexQueue = this.jobService.createQueue({
             name: 'update-search-index',
             process: job => {
                 const data = job.data;
                 switch (data.type) {
                     case 'reindex':
                         Logger.verbose(`sending ReindexMessage`);
-                        return this.sendMessageWithProgress(job, new ReindexMessage(data));
+                        return this.jobWithProgress(job, this.indexerController.reindex(data));
                     case 'update-product':
-                        return this.sendMessage(job, new UpdateProductMessage(data));
+                        return this.indexerController.updateProduct(data);
                     case 'update-variants':
-                        return this.sendMessage(job, new UpdateVariantMessage(data));
+                        return this.indexerController.updateVariants(data);
                     case 'delete-product':
-                        return this.sendMessage(job, new DeleteProductMessage(data));
+                        return this.indexerController.deleteProduct(data);
                     case 'delete-variant':
-                        return this.sendMessage(job, new DeleteVariantMessage(data));
+                        return this.indexerController.deleteVariants(data);
                     case 'update-variants-by-id':
-                        return this.sendMessageWithProgress(job, new UpdateVariantsByIdMessage(data));
+                        return this.jobWithProgress(job, this.indexerController.updateVariantsById(data));
                     case 'update-asset':
-                        return this.sendMessage(job, new UpdateAssetMessage(data));
+                        return this.indexerController.updateAsset(data);
                     case 'delete-asset':
-                        return this.sendMessage(job, new DeleteAssetMessage(data));
+                        return this.indexerController.deleteAsset(data);
                     case 'assign-product-to-channel':
-                        return this.sendMessage(job, new AssignProductToChannelMessage(data));
+                        return this.indexerController.assignProductToChannel(data);
                     case 'remove-product-from-channel':
-                        return this.sendMessage(job, new RemoveProductFromChannelMessage(data));
+                        return this.indexerController.removeProductFromChannel(data);
                     case 'assign-variant-to-channel':
-                        return this.sendMessage(job, new AssignVariantToChannelMessage(data));
+                        return this.indexerController.assignVariantToChannel(data);
                     case 'remove-variant-from-channel':
-                        return this.sendMessage(job, new RemoveVariantFromChannelMessage(data));
+                        return this.indexerController.removeVariantFromChannel(data);
                     default:
                         assertNever(data);
                         return Promise.resolve();
@@ -77,29 +65,29 @@ export class ElasticsearchIndexService {
     }
 
     reindex(ctx: RequestContext, dropIndices: boolean) {
-        return this.addJobToQueue({ type: 'reindex', ctx: ctx.serialize(), dropIndices });
+        return this.updateIndexQueue.add({ type: 'reindex', ctx: ctx.serialize(), dropIndices });
     }
 
     updateProduct(ctx: RequestContext, product: Product) {
-        this.addJobToQueue({ type: 'update-product', ctx: ctx.serialize(), productId: product.id });
+        this.updateIndexQueue.add({ type: 'update-product', ctx: ctx.serialize(), productId: product.id });
     }
 
     updateVariants(ctx: RequestContext, variants: ProductVariant[]) {
         const variantIds = variants.map(v => v.id);
-        this.addJobToQueue({ type: 'update-variants', ctx: ctx.serialize(), variantIds });
+        this.updateIndexQueue.add({ type: 'update-variants', ctx: ctx.serialize(), variantIds });
     }
 
     deleteProduct(ctx: RequestContext, product: Product) {
-        this.addJobToQueue({ type: 'delete-product', ctx: ctx.serialize(), productId: product.id });
+        this.updateIndexQueue.add({ type: 'delete-product', ctx: ctx.serialize(), productId: product.id });
     }
 
     deleteVariant(ctx: RequestContext, variants: ProductVariant[]) {
         const variantIds = variants.map(v => v.id);
-        this.addJobToQueue({ type: 'delete-variant', ctx: ctx.serialize(), variantIds });
+        this.updateIndexQueue.add({ type: 'delete-variant', ctx: ctx.serialize(), variantIds });
     }
 
     assignProductToChannel(ctx: RequestContext, product: Product, channelId: ID) {
-        this.addJobToQueue({
+        this.updateIndexQueue.add({
             type: 'assign-product-to-channel',
             ctx: ctx.serialize(),
             productId: product.id,
@@ -108,7 +96,7 @@ export class ElasticsearchIndexService {
     }
 
     removeProductFromChannel(ctx: RequestContext, product: Product, channelId: ID) {
-        this.addJobToQueue({
+        this.updateIndexQueue.add({
             type: 'remove-product-from-channel',
             ctx: ctx.serialize(),
             productId: product.id,
@@ -117,7 +105,7 @@ export class ElasticsearchIndexService {
     }
 
     assignVariantToChannel(ctx: RequestContext, productVariantId: ID, channelId: ID) {
-        this.addJobToQueue({
+        this.updateIndexQueue.add({
             type: 'assign-variant-to-channel',
             ctx: ctx.serialize(),
             productVariantId,
@@ -126,7 +114,7 @@ export class ElasticsearchIndexService {
     }
 
     removeVariantFromChannel(ctx: RequestContext, productVariantId: ID, channelId: ID) {
-        this.addJobToQueue({
+        this.updateIndexQueue.add({
             type: 'remove-variant-from-channel',
             ctx: ctx.serialize(),
             productVariantId,
@@ -135,41 +123,26 @@ export class ElasticsearchIndexService {
     }
 
     updateVariantsById(ctx: RequestContext, ids: ID[]) {
-        this.addJobToQueue({ type: 'update-variants-by-id', ctx: ctx.serialize(), ids });
+        this.updateIndexQueue.add({ type: 'update-variants-by-id', ctx: ctx.serialize(), ids });
     }
 
     updateAsset(ctx: RequestContext, asset: Asset) {
-        this.addJobToQueue({ type: 'update-asset', ctx: ctx.serialize(), asset: asset as any });
+        this.updateIndexQueue.add({ type: 'update-asset', ctx: ctx.serialize(), asset: asset as any });
     }
 
     deleteAsset(ctx: RequestContext, asset: Asset) {
-        this.addJobToQueue({ type: 'delete-asset', ctx: ctx.serialize(), asset: asset as any });
-    }
-
-    private addJobToQueue(data: UpdateIndexQueueJobData) {
-        if (updateIndexQueue) {
-            return updateIndexQueue.add(data);
-        }
-    }
-
-    private sendMessage(job: Job<any>, message: WorkerMessage<any, any>) {
-        return new Promise((resolve, reject) => {
-            this.workerService.send(message).subscribe({
-                complete: () => resolve(true),
-                error: err => {
-                    Logger.error(err);
-                    reject(err);
-                },
-            });
-        });
+        this.updateIndexQueue.add({ type: 'delete-asset', ctx: ctx.serialize(), asset: asset as any });
     }
 
-    private sendMessageWithProgress(job: Job<any>, message: WorkerMessage<any, ReindexMessageResponse>) {
+    private jobWithProgress(
+        job: Job<UpdateIndexQueueJobData>,
+        ob: Observable<ReindexMessageResponse>,
+    ): Promise<any> {
         return new Promise((resolve, reject) => {
             let total: number | undefined;
             let duration = 0;
             let completed = 0;
-            this.workerService.send(message).subscribe({
+            ob.subscribe({
                 next: (response: ReindexMessageResponse) => {
                     if (!total) {
                         total = response.total;

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

@@ -1,6 +1,5 @@
 import { Client } from '@elastic/elasticsearch';
-import { Controller, Inject, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
-import { MessagePattern } from '@nestjs/microservices';
+import { Inject, Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
 import { unique } from '@vendure/common/lib/unique';
 import {
     Asset,
@@ -30,22 +29,17 @@ import { ELASTIC_SEARCH_OPTIONS, loggerCtx, PRODUCT_INDEX_NAME, VARIANT_INDEX_NA
 import { createIndices, deleteByChannel, deleteIndices } from './indexing-utils';
 import { ElasticsearchOptions } from './options';
 import {
-    AssignProductToChannelMessage,
-    AssignVariantToChannelMessage,
     BulkOperation,
     BulkOperationDoc,
     BulkResponseBody,
-    DeleteAssetMessage,
-    DeleteProductMessage,
-    DeleteVariantMessage,
+    ProductChannelMessageData,
     ProductIndexItem,
-    ReindexMessage,
-    RemoveProductFromChannelMessage,
-    RemoveVariantFromChannelMessage,
-    UpdateAssetMessage,
-    UpdateProductMessage,
-    UpdateVariantMessage,
-    UpdateVariantsByIdMessage,
+    ReindexMessageData,
+    UpdateAssetMessageData,
+    UpdateProductMessageData,
+    UpdateVariantMessageData,
+    UpdateVariantsByIdMessageData,
+    VariantChannelMessageData,
     VariantIndexItem,
 } from './types';
 
@@ -70,7 +64,7 @@ export interface ReindexMessageResponse {
     duration: number;
 }
 
-@Controller()
+@Injectable()
 export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDestroy {
     private client: Client;
     private asyncQueue = new AsyncQueue('elasticsearch-indexer', 5);
@@ -96,160 +90,120 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
     /**
      * Updates the search index only for the affected product.
      */
-    @MessagePattern(UpdateProductMessage.pattern)
-    updateProduct({
-        ctx: rawContext,
-        productId,
-    }: UpdateProductMessage['data']): Observable<UpdateProductMessage['response']> {
+    async updateProduct({ ctx: rawContext, productId }: UpdateProductMessageData): Promise<boolean> {
         const ctx = RequestContext.deserialize(rawContext);
-        return asyncObservable(async () => {
-            await this.updateProductInternal(ctx, productId);
-            return true;
-        });
+        await this.updateProductInternal(ctx, productId);
+        return true;
     }
 
     /**
      * Updates the search index only for the affected product.
      */
-    @MessagePattern(DeleteProductMessage.pattern)
-    deleteProduct({
-        ctx: rawContext,
-        productId,
-    }: DeleteProductMessage['data']): Observable<DeleteProductMessage['response']> {
+    async deleteProduct({ ctx: rawContext, productId }: UpdateProductMessageData): Promise<boolean> {
         const ctx = RequestContext.deserialize(rawContext);
-        return asyncObservable(async () => {
-            const product = await this.connection.getRepository(Product).findOne(productId);
-            if (!product) {
-                return false;
-            }
-            await this.deleteProductInternal(product, ctx.channelId);
-            const variants = await this.productVariantService.getVariantsByProductId(ctx, productId);
-            await this.deleteVariantsInternal(variants, ctx.channelId);
-            return true;
-        });
+        const product = await this.connection.getRepository(Product).findOne(productId);
+        if (!product) {
+            return false;
+        }
+        await this.deleteProductInternal(product, ctx.channelId);
+        const variants = await this.productVariantService.getVariantsByProductId(ctx, productId);
+        await this.deleteVariantsInternal(variants, ctx.channelId);
+        return true;
     }
 
     /**
      * Updates the search index only for the affected product.
      */
-    @MessagePattern(AssignProductToChannelMessage.pattern)
-    assignProductsToChannel({
+    async assignProductToChannel({
         ctx: rawContext,
         productId,
         channelId,
-    }: AssignProductToChannelMessage['data']): Observable<AssignProductToChannelMessage['response']> {
+    }: ProductChannelMessageData): Promise<boolean> {
         const ctx = RequestContext.deserialize(rawContext);
-        return asyncObservable(async () => {
-            await this.updateProductInternal(ctx, productId);
-            const variants = await this.productVariantService.getVariantsByProductId(ctx, productId);
-            await this.updateVariantsInternal(
-                ctx,
-                variants.map(v => v.id),
-                channelId,
-            );
-            return true;
-        });
+        await this.updateProductInternal(ctx, productId);
+        const variants = await this.productVariantService.getVariantsByProductId(ctx, productId);
+        await this.updateVariantsInternal(
+            ctx,
+            variants.map(v => v.id),
+            channelId,
+        );
+        return true;
     }
 
     /**
      * Updates the search index only for the affected product.
      */
-    @MessagePattern(RemoveProductFromChannelMessage.pattern)
-    removeProductFromChannel({
+    async removeProductFromChannel({
         ctx: rawContext,
         productId,
         channelId,
-    }: RemoveProductFromChannelMessage['data']): Observable<RemoveProductFromChannelMessage['response']> {
+    }: ProductChannelMessageData): Promise<boolean> {
         const ctx = RequestContext.deserialize(rawContext);
-        return asyncObservable(async () => {
-            const product = await this.connection.getRepository(Product).findOne(productId);
-            if (!product) {
-                return false;
-            }
-            await this.deleteProductInternal(product, channelId);
-            const variants = await this.productVariantService.getVariantsByProductId(ctx, productId);
-            await this.deleteVariantsInternal(variants, channelId);
-            return true;
-        });
+        const product = await this.connection.getRepository(Product).findOne(productId);
+        if (!product) {
+            return false;
+        }
+        await this.deleteProductInternal(product, channelId);
+        const variants = await this.productVariantService.getVariantsByProductId(ctx, productId);
+        await this.deleteVariantsInternal(variants, channelId);
+        return true;
     }
 
-    @MessagePattern(AssignVariantToChannelMessage.pattern)
-    assignVariantToChannel({
+    async assignVariantToChannel({
         ctx: rawContext,
         productVariantId,
         channelId,
-    }: AssignVariantToChannelMessage['data']): Observable<AssignVariantToChannelMessage['response']> {
+    }: VariantChannelMessageData): Promise<boolean> {
         const ctx = RequestContext.deserialize(rawContext);
-        return asyncObservable(async () => {
-            await this.updateVariantsInternal(ctx, [productVariantId], channelId);
-            return true;
-        });
+        await this.updateVariantsInternal(ctx, [productVariantId], channelId);
+        return true;
     }
 
-    @MessagePattern(RemoveVariantFromChannelMessage.pattern)
-    removeVariantFromChannel({
+    async removeVariantFromChannel({
         ctx: rawContext,
         productVariantId,
         channelId,
-    }: RemoveVariantFromChannelMessage['data']): Observable<RemoveVariantFromChannelMessage['response']> {
+    }: VariantChannelMessageData): Promise<boolean> {
         const ctx = RequestContext.deserialize(rawContext);
-        return asyncObservable(async () => {
-            const productVariant = await this.connection.getEntityOrThrow(
-                ctx,
-                ProductVariant,
-                productVariantId,
-                { relations: ['product', 'product.channels'] },
-            );
-            await this.deleteVariantsInternal([productVariant], channelId);
-
-            if (!productVariant.product.channels.find(c => idsAreEqual(c.id, channelId))) {
-                await this.deleteProductInternal(productVariant.product, channelId);
-            }
-            return true;
+        const productVariant = await this.connection.getEntityOrThrow(ctx, ProductVariant, productVariantId, {
+            relations: ['product', 'product.channels'],
         });
+        await this.deleteVariantsInternal([productVariant], channelId);
+
+        if (!productVariant.product.channels.find(c => idsAreEqual(c.id, channelId))) {
+            await this.deleteProductInternal(productVariant.product, channelId);
+        }
+        return true;
     }
 
     /**
      * Updates the search index only for the affected entities.
      */
-    @MessagePattern(UpdateVariantMessage.pattern)
-    updateVariants({
-        ctx: rawContext,
-        variantIds,
-    }: UpdateVariantMessage['data']): Observable<UpdateVariantMessage['response']> {
+    async updateVariants({ ctx: rawContext, variantIds }: UpdateVariantMessageData): Promise<boolean> {
         const ctx = RequestContext.deserialize(rawContext);
-        return asyncObservable(async () => {
-            return this.asyncQueue.push(async () => {
-                await this.updateVariantsInternal(ctx, variantIds, ctx.channelId);
-                return true;
-            });
+        return this.asyncQueue.push(async () => {
+            await this.updateVariantsInternal(ctx, variantIds, ctx.channelId);
+            return true;
         });
     }
 
-    @MessagePattern(DeleteVariantMessage.pattern)
-    private deleteVaiants({
-        ctx: rawContext,
-        variantIds,
-    }: DeleteVariantMessage['data']): Observable<DeleteVariantMessage['response']> {
+    async deleteVariants({ ctx: rawContext, variantIds }: UpdateVariantMessageData): Promise<boolean> {
         const ctx = RequestContext.deserialize(rawContext);
-        return asyncObservable(async () => {
-            const variants = await this.connection
-                .getRepository(ProductVariant)
-                .findByIds(variantIds, { relations: ['product'] });
-            const productIds = unique(variants.map(v => v.product.id));
-            for (const productId of productIds) {
-                await this.updateProductInternal(ctx, productId);
-            }
-            await this.deleteVariantsInternal(variants, ctx.channelId);
-            return true;
-        });
+        const variants = await this.connection
+            .getRepository(ProductVariant)
+            .findByIds(variantIds, { relations: ['product'] });
+        const productIds = unique(variants.map(v => v.product.id));
+        for (const productId of productIds) {
+            await this.updateProductInternal(ctx, productId);
+        }
+        await this.deleteVariantsInternal(variants, ctx.channelId);
+        return true;
     }
 
-    @MessagePattern(UpdateVariantsByIdMessage.pattern)
     updateVariantsById({
         ctx: rawContext,
         ids,
-    }: UpdateVariantsByIdMessage['data']): Observable<UpdateVariantsByIdMessage['response']> {
+    }: UpdateVariantsByIdMessageData): Observable<ReindexMessageResponse> {
         const ctx = RequestContext.deserialize(rawContext);
         const { batchSize } = this.options;
 
@@ -326,11 +280,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
         });
     }
 
-    @MessagePattern(ReindexMessage.pattern)
-    reindex({
-        ctx: rawContext,
-        dropIndices,
-    }: ReindexMessage['data']): Observable<ReindexMessage['response']> {
+    reindex({ ctx: rawContext, dropIndices }: ReindexMessageData): Observable<ReindexMessageResponse> {
         const ctx = RequestContext.deserialize(rawContext);
         const { batchSize } = this.options;
 
@@ -401,34 +351,28 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
         });
     }
 
-    @MessagePattern(UpdateAssetMessage.pattern)
-    updateAsset(data: UpdateAssetMessage['data']): Observable<UpdateAssetMessage['response']> {
-        return asyncObservable(async () => {
-            const result1 = await this.updateAssetFocalPointForIndex(PRODUCT_INDEX_NAME, data.asset);
-            const result2 = await this.updateAssetFocalPointForIndex(VARIANT_INDEX_NAME, data.asset);
-            await this.client.indices.refresh({
-                index: [
-                    this.options.indexPrefix + PRODUCT_INDEX_NAME,
-                    this.options.indexPrefix + VARIANT_INDEX_NAME,
-                ],
-            });
-            return result1 && result2;
+    async updateAsset(data: UpdateAssetMessageData): Promise<boolean> {
+        const result1 = await this.updateAssetFocalPointForIndex(PRODUCT_INDEX_NAME, data.asset);
+        const result2 = await this.updateAssetFocalPointForIndex(VARIANT_INDEX_NAME, data.asset);
+        await this.client.indices.refresh({
+            index: [
+                this.options.indexPrefix + PRODUCT_INDEX_NAME,
+                this.options.indexPrefix + VARIANT_INDEX_NAME,
+            ],
         });
+        return result1 && result2;
     }
 
-    @MessagePattern(DeleteAssetMessage.pattern)
-    deleteAsset(data: DeleteAssetMessage['data']): Observable<DeleteAssetMessage['response']> {
-        return asyncObservable(async () => {
-            const result1 = await this.deleteAssetForIndex(PRODUCT_INDEX_NAME, data.asset);
-            const result2 = await this.deleteAssetForIndex(VARIANT_INDEX_NAME, data.asset);
-            await this.client.indices.refresh({
-                index: [
-                    this.options.indexPrefix + PRODUCT_INDEX_NAME,
-                    this.options.indexPrefix + VARIANT_INDEX_NAME,
-                ],
-            });
-            return result1 && result2;
+    async deleteAsset(data: UpdateAssetMessageData): Promise<boolean> {
+        const result1 = await this.deleteAssetForIndex(PRODUCT_INDEX_NAME, data.asset);
+        const result2 = await this.deleteAssetForIndex(VARIANT_INDEX_NAME, data.asset);
+        await this.client.indices.refresh({
+            index: [
+                this.options.indexPrefix + PRODUCT_INDEX_NAME,
+                this.options.indexPrefix + VARIANT_INDEX_NAME,
+            ],
         });
+        return result1 && result2;
     }
 
     private async updateAssetFocalPointForIndex(indexName: string, asset: Asset): Promise<boolean> {

packages/elasticsearch-plugin/src/plugin.ts

@@ -196,6 +196,7 @@ import { ElasticsearchOptions, ElasticsearchRuntimeOptions, mergeWithDefaults }
         ElasticsearchIndexService,
         ElasticsearchService,
         ElasticsearchHealthIndicator,
+        ElasticsearchIndexerController,
         { provide: ELASTIC_SEARCH_OPTIONS, useFactory: () => ElasticsearchPlugin.options },
     ],
     adminApiExtensions: { resolvers: [AdminElasticSearchResolver] },
@@ -213,7 +214,6 @@ import { ElasticsearchOptions, ElasticsearchRuntimeOptions, mergeWithDefaults }
         // which looks like possibly a TS/definitions bug.
         schema: () => generateSchemaExtensions(ElasticsearchPlugin.options as any),
     },
-    workers: [ElasticsearchIndexerController],
 })
 export class ElasticsearchPlugin implements OnApplicationBootstrap {
     private static options: ElasticsearchRuntimeOptions;
@@ -252,7 +252,6 @@ export class ElasticsearchPlugin implements OnApplicationBootstrap {
         Logger.info(`Successfully connected to Elasticsearch instance at "${nodeName}"`, loggerCtx);
 
         await this.elasticsearchService.createIndicesIfNotExists();
-        this.elasticsearchIndexService.initJobQueue();
         this.healthCheckRegistryService.registerIndicatorFunction(() =>
             this.elasticsearchHealthIndicator.isHealthy(),
         );

packages/elasticsearch-plugin/src/types.ts

@@ -8,7 +8,7 @@ import {
     SearchResult,
 } from '@vendure/common/lib/generated-types';
 import { ID, JsonCompatible } from '@vendure/common/lib/shared-types';
-import { Asset, SerializedRequestContext, WorkerMessage } from '@vendure/core';
+import { Asset, SerializedRequestContext } from '@vendure/core';
 
 export type ElasticSearchInput = SearchInput & {
     priceRange?: PriceRange;
@@ -190,46 +190,6 @@ export interface UpdateAssetMessageData {
     asset: JsonCompatible<Required<Asset>>;
 }
 
-export class ReindexMessage extends WorkerMessage<ReindexMessageData, ReindexMessageResponse> {
-    static readonly pattern = 'Reindex';
-}
-export class UpdateVariantMessage extends WorkerMessage<UpdateVariantMessageData, boolean> {
-    static readonly pattern = 'UpdateProduct';
-}
-export class UpdateProductMessage extends WorkerMessage<UpdateProductMessageData, boolean> {
-    static readonly pattern = 'UpdateVariant';
-}
-export class DeleteVariantMessage extends WorkerMessage<UpdateVariantMessageData, boolean> {
-    static readonly pattern = 'DeleteProduct';
-}
-export class DeleteProductMessage extends WorkerMessage<UpdateProductMessageData, boolean> {
-    static readonly pattern = 'DeleteVariant';
-}
-export class UpdateVariantsByIdMessage extends WorkerMessage<
-    UpdateVariantsByIdMessageData,
-    ReindexMessageResponse
-> {
-    static readonly pattern = 'UpdateVariantsById';
-}
-export class AssignProductToChannelMessage extends WorkerMessage<ProductChannelMessageData, boolean> {
-    static readonly pattern = 'AssignProductToChannel';
-}
-export class RemoveProductFromChannelMessage extends WorkerMessage<ProductChannelMessageData, boolean> {
-    static readonly pattern = 'RemoveProductFromChannel';
-}
-export class AssignVariantToChannelMessage extends WorkerMessage<VariantChannelMessageData, boolean> {
-    static readonly pattern = 'AssignVariantToChannel';
-}
-export class RemoveVariantFromChannelMessage extends WorkerMessage<VariantChannelMessageData, boolean> {
-    static readonly pattern = 'RemoveVariantFromChannel';
-}
-export class UpdateAssetMessage extends WorkerMessage<UpdateAssetMessageData, boolean> {
-    static readonly pattern = 'UpdateAsset';
-}
-export class DeleteAssetMessage extends WorkerMessage<UpdateAssetMessageData, boolean> {
-    static readonly pattern = 'DeleteAsset';
-}
-
 type Maybe<T> = T | undefined;
 type CustomMappingDefinition<Args extends any[], T extends string, R> = {
     graphQlType: T;