feat(elasticsearch-plugin): Update index on asset deletion

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

@@ -20,6 +20,7 @@ import {
     CreateCollection,
     CreateFacet,
     CurrencyCode,
+    DeleteAsset,
     DeleteProduct,
     DeleteProductVariant,
     LanguageCode,
@@ -39,6 +40,7 @@ import {
     CREATE_CHANNEL,
     CREATE_COLLECTION,
     CREATE_FACET,
+    DELETE_ASSET,
     DELETE_PRODUCT,
     DELETE_PRODUCT_VARIANT,
     REMOVE_PRODUCT_FROM_CHANNEL,
@@ -611,42 +613,59 @@ describe('Elasticsearch plugin', () => {
                 ]);
             });
 
-            it('updates index when asset focalPoint is changed', async () => {
-                const { search: search1 } = await doAdminSearchQuery({
-                    term: 'laptop',
-                    groupByProduct: true,
-                    take: 1,
-                    sort: {
-                        name: SortOrder.ASC,
-                    },
-                });
+            describe('asset changes', () => {
+                function searchForLaptop() {
+                    return doAdminSearchQuery({
+                        term: 'laptop',
+                        groupByProduct: true,
+                        take: 1,
+                        sort: {
+                            name: SortOrder.ASC,
+                        },
+                    });
+                }
 
-                expect(search1.items[0].productAsset!.id).toBe('T_1');
-                expect(search1.items[0].productAsset!.focalPoint).toBeNull();
+                it('updates index when asset focalPoint is changed', async () => {
+                    const { search: search1 } = await searchForLaptop();
 
-                await adminClient.query<UpdateAsset.Mutation, UpdateAsset.Variables>(UPDATE_ASSET, {
-                    input: {
-                        id: 'T_1',
-                        focalPoint: {
-                            x: 0.42,
-                            y: 0.42,
+                    expect(search1.items[0].productAsset!.id).toBe('T_1');
+                    expect(search1.items[0].productAsset!.focalPoint).toBeNull();
+
+                    await adminClient.query<UpdateAsset.Mutation, UpdateAsset.Variables>(UPDATE_ASSET, {
+                        input: {
+                            id: 'T_1',
+                            focalPoint: {
+                                x: 0.42,
+                                y: 0.42,
+                            },
                         },
-                    },
-                });
+                    });
 
-                await awaitRunningJobs(adminClient);
+                    await awaitRunningJobs(adminClient);
 
-                const { search: search2 } = await doAdminSearchQuery({
-                    term: 'laptop',
-                    groupByProduct: true,
-                    take: 1,
-                    sort: {
-                        name: SortOrder.ASC,
-                    },
+                    const { search: search2 } = await searchForLaptop();
+
+                    expect(search2.items[0].productAsset!.id).toBe('T_1');
+                    expect(search2.items[0].productAsset!.focalPoint).toEqual({ x: 0.42, y: 0.42 });
                 });
 
-                expect(search2.items[0].productAsset!.id).toBe('T_1');
-                expect(search2.items[0].productAsset!.focalPoint).toEqual({ x: 0.42, y: 0.42 });
+                it('updates index when asset deleted', async () => {
+                    const { search: search1 } = await searchForLaptop();
+
+                    const assetId = search1.items[0].productAsset?.id;
+                    expect(assetId).toBeTruthy();
+
+                    await adminClient.query<DeleteAsset.Mutation, DeleteAsset.Variables>(DELETE_ASSET, {
+                        id: assetId!,
+                        force: true,
+                    });
+
+                    await awaitRunningJobs(adminClient);
+
+                    const { search: search2 } = await searchForLaptop();
+
+                    expect(search2.items[0].productAsset).toBeNull();
+                });
             });
 
             it('does not include deleted ProductVariants in index', async () => {

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

@@ -1,4 +1,5 @@
 import { Injectable } from '@nestjs/common';
+import { assertNever } from '@vendure/common/lib/shared-utils';
 import {
     Asset,
     ID,
@@ -16,6 +17,7 @@ import {
 import { ReindexMessageResponse } from './indexer.controller';
 import {
     AssignProductToChannelMessage,
+    DeleteAssetMessage,
     DeleteProductMessage,
     DeleteVariantMessage,
     ReindexMessage,
@@ -62,12 +64,17 @@ export class ElasticsearchIndexService {
                     case 'update-asset':
                         this.sendMessage(job, new UpdateAssetMessage(data));
                         break;
+                    case 'delete-asset':
+                        this.sendMessage(job, new DeleteAssetMessage(data));
+                        break;
                     case 'assign-product-to-channel':
                         this.sendMessage(job, new AssignProductToChannelMessage(data));
                         break;
                     case 'remove-product-from-channel':
                         this.sendMessage(job, new RemoveProductFromChannelMessage(data));
                         break;
+                    default:
+                        assertNever(data);
                 }
             },
         });
@@ -121,6 +128,10 @@ export class ElasticsearchIndexService {
         this.addJobToQueue({ 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);

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

@@ -35,6 +35,7 @@ import {
     BulkOperation,
     BulkOperationDoc,
     BulkResponseBody,
+    DeleteAssetMessage,
     DeleteProductMessage,
     DeleteVariantMessage,
     ProductIndexItem,
@@ -325,8 +326,8 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
     @MessagePattern(UpdateAssetMessage.pattern)
     updateAsset(data: UpdateAssetMessage['data']): Observable<UpdateAssetMessage['response']> {
         return asyncObservable(async () => {
-            const result1 = await this.updateAssetForIndex(PRODUCT_INDEX_NAME, data.asset);
-            const result2 = await this.updateAssetForIndex(VARIANT_INDEX_NAME, data.asset);
+            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,
@@ -337,16 +338,57 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
         });
     }
 
-    private async updateAssetForIndex(indexName: string, asset: Asset): Promise<boolean> {
+    @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;
+        });
+    }
+
+    private async updateAssetFocalPointForIndex(indexName: string, asset: Asset): Promise<boolean> {
         const focalPoint = asset.focalPoint || null;
         const params = { focalPoint };
+        return this.updateAssetForIndex(
+            indexName,
+            asset,
+            {
+                source: 'ctx._source.productPreviewFocalPoint = params.focalPoint',
+                params,
+            },
+            {
+                source: 'ctx._source.productVariantPreviewFocalPoint = params.focalPoint',
+                params,
+            },
+        );
+    }
+
+    private async deleteAssetForIndex(indexName: string, asset: Asset): Promise<boolean> {
+        return this.updateAssetForIndex(
+            indexName,
+            asset,
+            { source: 'ctx._source.productAssetId = null' },
+            { source: 'ctx._source.productVariantAssetId = null' },
+        );
+    }
+
+    private async updateAssetForIndex(
+        indexName: string,
+        asset: Asset,
+        updateProductScript: { source: string; params?: any },
+        updateVariantScript: { source: string; params?: any },
+    ): Promise<boolean> {
         const result1 = await this.client.update_by_query({
             index: this.options.indexPrefix + indexName,
             body: {
-                script: {
-                    source: 'ctx._source.productPreviewFocalPoint = params.focalPoint',
-                    params,
-                },
+                script: updateProductScript,
                 query: {
                     term: {
                         productAssetId: asset.id,
@@ -360,10 +402,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
         const result2 = await this.client.update_by_query({
             index: this.options.indexPrefix + indexName,
             body: {
-                script: {
-                    source: 'ctx._source.productVariantPreviewFocalPoint = params.focalPoint',
-                    params,
-                },
+                script: updateVariantScript,
                 query: {
                     term: {
                         productVariantAssetId: asset.id,

packages/elasticsearch-plugin/src/plugin.ts

@@ -260,6 +260,9 @@ export class ElasticsearchPlugin implements OnVendureBootstrap {
             if (event.type === 'updated') {
                 return this.elasticsearchIndexService.updateAsset(event.ctx, event.asset);
             }
+            if (event.type === 'deleted') {
+                return this.elasticsearchIndexService.deleteAsset(event.ctx, event.asset);
+            }
         });
 
         this.eventBus.ofType(ProductChannelEvent).subscribe((event) => {

packages/elasticsearch-plugin/src/types.ts

@@ -210,6 +210,9 @@ export class RemoveProductFromChannelMessage extends WorkerMessage<ProductChanne
 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 | null | undefined;
 type CustomMappingDefinition<Args extends any[], T extends string, R> = {
@@ -226,6 +229,7 @@ type DeleteProductJobData = NamedJobData<'delete-product', UpdateProductMessageD
 type DeleteVariantJobData = NamedJobData<'delete-variant', UpdateVariantMessageData>;
 type UpdateVariantsByIdJobData = NamedJobData<'update-variants-by-id', UpdateVariantsByIdMessageData>;
 type UpdateAssetJobData = NamedJobData<'update-asset', UpdateAssetMessageData>;
+type DeleteAssetJobData = NamedJobData<'delete-asset', UpdateAssetMessageData>;
 type AssignProductToChannelJobData = NamedJobData<'assign-product-to-channel', ProductChannelMessageData>;
 type RemoveProductFromChannelJobData = NamedJobData<'remove-product-from-channel', ProductChannelMessageData>;
 export type UpdateIndexQueueJobData =
@@ -236,6 +240,7 @@ export type UpdateIndexQueueJobData =
     | DeleteVariantJobData
     | UpdateVariantsByIdJobData
     | UpdateAssetJobData
+    | DeleteAssetJobData
     | AssignProductToChannelJobData
     | RemoveProductFromChannelJobData;