perf(elasticsearch): added optimization for each opirations during reindex and other operations

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

@@ -1,18 +1,18 @@
-import { Client } from '@elastic/elasticsearch';
-import { Inject, Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import type { Client } from '@elastic/elasticsearch';
+import type { OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { Inject, Injectable } from '@nestjs/common';
 import { unique } from '@vendure/common/lib/unique';
 import {
     Asset,
-    asyncObservable,
-    AsyncQueue,
-    Channel,
     Collection,
-    ConfigService,
     EntityRelationPaths,
     FacetValue,
     ID,
-    InternalServerError,
     LanguageCode,
+    AsyncQueue,
+    Channel,
+    ConfigService,
+    InternalServerError,
     Logger,
     Product,
     ProductPriceApplicator,
@@ -27,9 +27,9 @@ import {
 import { Observable } from 'rxjs';
 import { In, IsNull } from 'typeorm';
 
-import { ELASTIC_SEARCH_OPTIONS, loggerCtx, VARIANT_INDEX_NAME } from '../constants';
-import { ElasticsearchOptions } from '../options';
-import {
+import { ELASTIC_SEARCH_OPTIONS, VARIANT_INDEX_NAME, loggerCtx } from '../constants';
+import type { ElasticsearchOptions } from '../options';
+import type {
     BulkOperation,
     BulkOperationDoc,
     BulkResponseBody,
@@ -46,6 +46,7 @@ import {
 
 import { createIndices, getClient, getIndexNameByAlias } from './indexing-utils';
 import { MutableRequestContext } from './mutable-request-context';
+import { asyncObservable } from "@vendure/core/src";
 
 export const defaultProductRelations: Array<EntityRelationPaths<Product>> = [
     'featuredAsset',
@@ -111,7 +112,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
     /**
      * Updates the search index only for the affected product.
      */
-    async updateProduct({ ctx: rawContext, productId }: UpdateProductMessageData): Promise<boolean> {
+    async updateProduct({ctx: rawContext, productId}: UpdateProductMessageData): Promise<boolean> {
         const ctx = MutableRequestContext.deserialize(rawContext);
         await this.updateProductsInternal(ctx, [productId]);
         return true;
@@ -120,12 +121,11 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
     /**
      * Updates the search index only for the affected product.
      */
-    async deleteProduct({ ctx: rawContext, productId }: UpdateProductMessageData): Promise<boolean> {
-        const operations = await this.deleteProductOperations(
+    async deleteProduct({ctx: rawContext, productId}: UpdateProductMessageData): Promise<boolean> {
+        await this.deleteProductOperations(
             RequestContext.deserialize(rawContext),
             productId,
         );
-        await this.executeBulkOperations(operations);
         return true;
     }
 
@@ -180,7 +180,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
     /**
      * Updates the search index only for the affected entities.
      */
-    async updateVariants({ ctx: rawContext, variantIds }: UpdateVariantMessageData): Promise<boolean> {
+    async updateVariants({ctx: rawContext, variantIds}: UpdateVariantMessageData): Promise<boolean> {
         const ctx = MutableRequestContext.deserialize(rawContext);
         return this.asyncQueue.push(async () => {
             const productIds = await this.getProductIdsByVariantIds(variantIds);
@@ -189,7 +189,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
         });
     }
 
-    async deleteVariants({ ctx: rawContext, variantIds }: UpdateVariantMessageData): Promise<boolean> {
+    async deleteVariants({ctx: rawContext, variantIds}: UpdateVariantMessageData): Promise<boolean> {
         const ctx = MutableRequestContext.deserialize(rawContext);
         const productIds = await this.getProductIdsByVariantIds(variantIds);
         for (const productId of productIds) {
@@ -236,9 +236,9 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                 const ctx = MutableRequestContext.deserialize(rawContext);
 
                 const reindexTempName = new Date().getTime();
-                const variantIndexName = this.options.indexPrefix + VARIANT_INDEX_NAME;
-                const variantIndexNameForReindex = VARIANT_INDEX_NAME + `-reindex-${reindexTempName}`;
-                const reindexVariantAliasName = this.options.indexPrefix + variantIndexNameForReindex;
+                const variantIndexName = `${this.options.indexPrefix}${VARIANT_INDEX_NAME}`;
+                const variantIndexNameForReindex = `${VARIANT_INDEX_NAME}-reindex-${reindexTempName}`;
+                const reindexVariantAliasName = `${this.options.indexPrefix}${variantIndexNameForReindex}`;
                 try {
                     await createIndices(
                         this.client,
@@ -267,8 +267,6 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                 let skip = 0;
                 let finishedProductsCount = 0;
                 do {
-                    let operations: BulkVariantOperation[] = [];
-
                     productIds = await this.connection
                         .rawConnection
                         .getRepository(Product)
@@ -279,18 +277,8 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                         .take(this.options.reindexProductsChunkSize)
                         .getMany();
 
-                    for (const { id: productId } of productIds) {
-                        operations.push(...(await this.updateProductsOperationsOnly(ctx, productId)));
-                        if (operations.length >= this.options.reindexBulkOperationSizeLimit) {
-                            // Because we can have a huge amount of variant for 1 product, we also chunk update operations
-                            await this.executeBulkOperationsByChunks(
-                                this.options.reindexBulkOperationSizeLimit,
-                                operations,
-                                variantIndexNameForReindex,
-                            );
-                            // Reset operations to avoid memory peaks with huge amount of operations
-                            operations = [];
-                        }
+                    for (const {id: productId} of productIds) {
+                        await this.updateProductsOperationsOnly(ctx, productId, variantIndexNameForReindex);
                         finishedProductsCount++;
                         observer.next({
                             total: totalProductIds,
@@ -299,98 +287,13 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                         });
                     }
 
-                    Logger.verbose(`Will execute ${operations.length} bulk update operations`, loggerCtx);
-
-                    // Because we can have a huge amount of variant for 1 product, we also chunk update operations
-                    await this.executeBulkOperationsByChunks(
-                        this.options.reindexBulkOperationSizeLimit,
-                        operations,
-                        variantIndexNameForReindex,
-                    );
-
                     skip += this.options.reindexProductsChunkSize;
 
                     Logger.verbose(`Done ${finishedProductsCount} / ${totalProductIds} products`);
-                } while (productIds.length >= this.options.reindexProductsChunkSize);
+                } while (productIds.length >= this.options.reindexProductsChunkSize)
 
                 // Switch the index to the new reindexed one
-                try {
-                    const reindexVariantAliasExist = await this.client.indices.existsAlias({
-                        name: reindexVariantAliasName,
-                    });
-                    if (reindexVariantAliasExist) {
-                        const reindexVariantIndexName = await getIndexNameByAlias(
-                            this.client,
-                            reindexVariantAliasName,
-                        );
-                        const originalVariantAliasExist = await this.client.indices.existsAlias({
-                            name: variantIndexName,
-                        });
-                        const originalVariantIndexExist = await this.client.indices.exists({
-                            index: variantIndexName,
-                        });
-
-                        const originalVariantIndexName = await getIndexNameByAlias(
-                            this.client,
-                            variantIndexName,
-                        );
-
-                        const actions = [
-                            {
-                                remove: {
-                                    index: reindexVariantIndexName,
-                                    alias: reindexVariantAliasName,
-                                },
-                            },
-                            {
-                                add: {
-                                    index: reindexVariantIndexName,
-                                    alias: variantIndexName,
-                                },
-                            },
-                        ];
-
-                        if (originalVariantAliasExist.body) {
-                            actions.push({
-                                remove: {
-                                    index: originalVariantIndexName,
-                                    alias: variantIndexName,
-                                },
-                            });
-                        } else if (originalVariantIndexExist.body) {
-                            await this.client.indices.delete({
-                                index: [variantIndexName],
-                            });
-                        }
-
-                        await this.client.indices.updateAliases({
-                            body: {
-                                actions,
-                            },
-                        });
-
-                        if (originalVariantAliasExist.body) {
-                            await this.client.indices.delete({
-                                index: [originalVariantIndexName],
-                            });
-                        }
-                    }
-                } catch (e: any) {
-                    Logger.error('Could not switch indexes');
-                } finally {
-                    const reindexVariantAliasExist = await this.client.indices.existsAlias({
-                        name: reindexVariantAliasName,
-                    });
-                    if (reindexVariantAliasExist.body) {
-                        const reindexVariantAliasResult = await this.client.indices.getAlias({
-                            name: reindexVariantAliasName,
-                        });
-                        const reindexVariantIndexName = Object.keys(reindexVariantAliasResult.body)[0];
-                        await this.client.indices.delete({
-                            index: [reindexVariantIndexName],
-                        });
-                    }
-                }
+                await this.switchAlias(reindexVariantAliasName, variantIndexName);
 
                 Logger.verbose('Completed reindexing!', loggerCtx);
 
@@ -400,7 +303,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                     duration: +new Date() - timeStart,
                 };
             });
-        });
+        })
     }
 
     async executeBulkOperationsByChunks(
@@ -408,6 +311,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
         operations: BulkVariantOperation[],
         index = VARIANT_INDEX_NAME,
     ): Promise<void> {
+        Logger.verbose(`Will execute ${operations.length} bulk update operations with index ${index}`, loggerCtx);
         let i;
         let j;
         let processedOperation = 0;
@@ -441,7 +345,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
 
     private async updateAssetFocalPointForIndex(indexName: string, asset: Asset): Promise<boolean> {
         const focalPoint = asset.focalPoint || null;
-        const params = { focalPoint };
+        const params = {focalPoint};
         return this.updateAssetForIndex(
             indexName,
             asset,
@@ -482,9 +386,9 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                 },
             },
         });
-        for (const failure of result1.body.failures) {
+        for (const failure of result1.body.failures)
             Logger.error(`${failure.cause.type as string}: ${failure.cause.reason as string}`, loggerCtx);
-        }
+
         const result2 = await this.client.update_by_query({
             index: this.options.indexPrefix + indexName,
             body: {
@@ -503,32 +407,117 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
     }
 
     private async updateProductsInternal(ctx: MutableRequestContext, productIds: ID[]) {
-        const operations = await this.updateProductsOperations(ctx, productIds);
-        await this.executeBulkOperations(operations);
+        await this.updateProductsOperations(ctx, productIds);
+    }
+
+    private async switchAlias(
+        reindexVariantAliasName: string,
+        variantIndexName: string,
+    ): Promise<void> {
+        try {
+            const reindexVariantAliasExist = await this.client.indices.existsAlias({
+                name: reindexVariantAliasName,
+            });
+            if (reindexVariantAliasExist) {
+                const reindexVariantIndexName = await getIndexNameByAlias(
+                    this.client,
+                    reindexVariantAliasName,
+                );
+                const originalVariantAliasExist = await this.client.indices.existsAlias({
+                    name: variantIndexName,
+                });
+                const originalVariantIndexExist = await this.client.indices.exists({
+                    index: variantIndexName,
+                });
+
+                const originalVariantIndexName = await getIndexNameByAlias(
+                    this.client,
+                    variantIndexName,
+                );
+
+                const actions = [
+                    {
+                        remove: {
+                            index: reindexVariantIndexName,
+                            alias: reindexVariantAliasName,
+                        },
+                    },
+                    {
+                        add: {
+                            index: reindexVariantIndexName,
+                            alias: variantIndexName,
+                        },
+                    },
+                ];
+
+                if (originalVariantAliasExist.body) {
+                    actions.push({
+                        remove: {
+                            index: originalVariantIndexName,
+                            alias: variantIndexName,
+                        },
+                    });
+                } else if (originalVariantIndexExist.body) {
+                    await this.client.indices.delete({
+                        index: [variantIndexName],
+                    });
+                }
+
+                await this.client.indices.updateAliases({
+                    body: {
+                        actions,
+                    },
+                });
+
+                if (originalVariantAliasExist.body) {
+                    await this.client.indices.delete({
+                        index: [originalVariantIndexName],
+                    });
+                }
+            }
+        } catch (e: any) {
+            Logger.error('Could not switch indexes');
+        } finally {
+            const reindexVariantAliasExist = await this.client.indices.existsAlias({
+                name: reindexVariantAliasName,
+            });
+            if (reindexVariantAliasExist.body) {
+                const reindexVariantAliasResult = await this.client.indices.getAlias({
+                    name: reindexVariantAliasName,
+                });
+                const reindexVariantIndexName = Object.keys(reindexVariantAliasResult.body)[0]
+                await this.client.indices.delete({
+                    index: [reindexVariantIndexName],
+                });
+            }
+        }
     }
 
     private async updateProductsOperationsOnly(
         ctx: MutableRequestContext,
         productId: ID,
-    ): Promise<BulkVariantOperation[]> {
-        const operations: BulkVariantOperation[] = [];
+        index = VARIANT_INDEX_NAME
+    ): Promise<void> {
+        let operations: BulkVariantOperation[] = [];
         let product: Product | undefined;
         try {
             product = await this.connection
+                .rawConnection
                 .getRepository(Product)
-                .findOne({
-                    where: { id: productId, deletedAt: IsNull() },
+                .find({
+                    relationLoadStrategy: 'query',
+                    where: {id: productId, deletedAt: IsNull()},
                     relations: this.productRelations,
                 })
-                .then(result => result ?? undefined);
+                .then(result => result[0] ?? undefined);
         } catch (e: any) {
             Logger.error(e.message, loggerCtx, e.stack);
             throw e;
         }
-        if (!product) {
-            return operations;
-        }
-        const updatedProductVariants = await this.connection.getRepository(ProductVariant).find({
+        if (!product)
+            return;
+
+        const updatedProductVariants = await this.connection.rawConnection.getRepository(ProductVariant).find({
             relations: this.variantRelations,
             where: {
                 productId,
@@ -537,27 +526,28 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
             order: {
                 id: 'ASC',
             },
+            relationLoadStrategy: 'query',
         });
-        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+
         updatedProductVariants.forEach(variant => (variant.product = product!));
-        if (!product.enabled) {
+        if (!product.enabled)
             updatedProductVariants.forEach(v => (v.enabled = false));
-        }
-        Logger.debug(`Updating Product (${productId})`, loggerCtx);
-        const languageVariants: LanguageCode[] = [];
+
+        Logger.debug(`Updating Product (${productId})`, loggerCtx)
+        const languageVariants: LanguageCode[] = []
         languageVariants.push(...product.translations.map(t => t.languageCode));
-        for (const variant of updatedProductVariants) {
+        for (const variant of updatedProductVariants)
             languageVariants.push(...variant.translations.map(t => t.languageCode));
-        }
-        const uniqueLanguageVariants = unique(languageVariants);
+
+        const uniqueLanguageVariants = unique(languageVariants)
         for (const channel of product.channels) {
             ctx.setChannel(channel);
             const variantsInChannel = updatedProductVariants.filter(v =>
                 v.channels.map(c => c.id).includes(ctx.channelId),
             );
-            for (const variant of variantsInChannel) {
+            for (const variant of variantsInChannel)
                 await this.productPriceApplicator.applyChannelPriceAndTax(variant, ctx);
-            }
+
             for (const languageCode of uniqueLanguageVariants) {
                 if (variantsInChannel.length) {
                     for (const variant of variantsInChannel) {
@@ -586,7 +576,17 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                                     doc_as_upsert: true,
                                 },
                             },
-                        );
+                        )
+
+                        if (operations.length >= this.options.reindexBulkOperationSizeLimit) {
+                            // Because we can have a huge amount of variant for 1 product, we also chunk update operations
+                            await this.executeBulkOperationsByChunks(
+                                this.options.reindexBulkOperationSizeLimit,
+                                operations,
+                                index,
+                            );
+                            operations = []
+                        }
                     }
                 } else {
                     operations.push(
@@ -605,30 +605,44 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                         {
                             index: VARIANT_INDEX_NAME,
                             operation: {
-                                doc: this.createSyntheticProductIndexItem(product, ctx, languageCode),
+                                doc: await this.createSyntheticProductIndexItem(product, ctx, languageCode),
                                 doc_as_upsert: true,
                             },
                         },
                     );
                 }
+                if (operations.length >= this.options.reindexBulkOperationSizeLimit) {
+                    // Because we can have a huge amount of variant for 1 product, we also chunk update operations
+                    await this.executeBulkOperationsByChunks(
+                        this.options.reindexBulkOperationSizeLimit,
+                        operations,
+                        index,
+                    );
+                    operations = []
+                }
             }
         }
 
-        return operations;
+        // Because we can have a huge amount of variant for 1 product, we also chunk update operations
+        await this.executeBulkOperationsByChunks(
+            this.options.reindexBulkOperationSizeLimit,
+            operations,
+            index,
+        );
+
+        return;
     }
 
     private async updateProductsOperations(
         ctx: MutableRequestContext,
         productIds: ID[],
-    ): Promise<BulkVariantOperation[]> {
-        Logger.debug(`Updating ${productIds.length} Products`, loggerCtx);
-        const operations: BulkVariantOperation[] = [];
-
+    ): Promise<void> {
+        Logger.debug(`Updating ${productIds.length} Products`, loggerCtx)
         for (const productId of productIds) {
-            operations.push(...(await this.deleteProductOperations(ctx, productId)));
-            operations.push(...(await this.updateProductsOperationsOnly(ctx, productId)));
+            await this.deleteProductOperations(ctx, productId);
+            await this.updateProductsOperationsOnly(ctx, productId);
         }
-        return operations;
+        return;
     }
 
     /**
@@ -649,7 +663,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
         for (const relation of hydratedRelations) {
             let path = relation.split('.');
             if (path[0] === 'customFields') {
-                if (2 < path.length) {
+                if (path.length > 2) {
                     throw new InternalServerError(
                         [
                             'hydrateProductRelations / hydrateProductVariantRelations does not currently support nested custom field relations',
@@ -674,7 +688,8 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
     private async deleteProductOperations(
         ctx: RequestContext,
         productId: ID,
-    ): Promise<BulkVariantOperation[]> {
+        index: string = VARIANT_INDEX_NAME,
+    ): Promise<void> {
         const channels = await this.requestContextCache.get(ctx, 'elastic-index-all-channels', () =>
             this.connection
                 .rawConnection
@@ -701,20 +716,20 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
             .andWhere('channel.id = :channelId', { channelId: ctx.channelId })
             .getOne();
 
-        if (!product) {
-            return [];
-        }
+        if (!product)
+            return;
 
         Logger.debug(`Deleting 1 Product (id: ${productId})`, loggerCtx);
-        const operations: BulkVariantOperation[] = [];
+        let operations: BulkVariantOperation[] = [];
         const languageVariants: LanguageCode[] = [];
         languageVariants.push(...product.translations.map(t => t.languageCode));
-        for (const variant of product.variants) {
+        for (const variant of product.variants)
             languageVariants.push(...variant.translations.map(t => t.languageCode));
-        }
+
         const uniqueLanguageVariants = unique(languageVariants);
 
-        for (const { id: channelId } of channels) {
+
+        for (const {id: channelId} of channels) {
             for (const languageCode of uniqueLanguageVariants) {
                 operations.push({
                     index: VARIANT_INDEX_NAME,
@@ -724,25 +739,42 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                         },
                     },
                 });
+                if (operations.length >= this.options.reindexBulkOperationSizeLimit) {
+                    // Because we can have a huge amount of variant for 1 product, we also chunk update operations
+                    await this.executeBulkOperationsByChunks(
+                        this.options.reindexBulkOperationSizeLimit,
+                        operations,
+                        index,
+                    );
+                    operations = []
+                }
             }
         }
-        operations.push(
-            ...(await this.deleteVariantsInternalOperations(
-                product.variants,
-                channels.map(c => c.id),
-                uniqueLanguageVariants,
-            )),
+        // Because we can have a huge amount of variant for 1 product, we also chunk update operations
+        await this.executeBulkOperationsByChunks(
+            this.options.reindexBulkOperationSizeLimit,
+            operations,
+            index,
+        );
+
+        await this.deleteVariantsInternalOperations(
+            product.variants,
+            channels.map(c => c.id),
+            uniqueLanguageVariants,
+            index,
         );
-        return operations;
+
+        return;
     }
 
     private async deleteVariantsInternalOperations(
         variants: ProductVariant[],
         channelIds: ID[],
         languageVariants: LanguageCode[],
-    ): Promise<BulkVariantOperation[]> {
-        Logger.debug(`Deleting ${variants.length} ProductVariants`, loggerCtx);
-        const operations: BulkVariantOperation[] = [];
+        index = VARIANT_INDEX_NAME,
+    ): Promise<void> {
+        Logger.debug(`Deleting ${variants.length} ProductVariants`, loggerCtx)
+        let operations: BulkVariantOperation[] = [];
         for (const variant of variants) {
             for (const channelId of channelIds) {
                 for (const languageCode of languageVariants) {
@@ -757,11 +789,26 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                                 ),
                             },
                         },
-                    });
+                    })
+                    if (operations.length >= this.options.reindexBulkOperationSizeLimit) {
+                        // Because we can have a huge amount of variant for 1 product, we also chunk update operations
+                        await this.executeBulkOperationsByChunks(
+                            this.options.reindexBulkOperationSizeLimit,
+                            operations,
+                            index,
+                        )
+                        operations = []
+                    }
                 }
             }
         }
-        return operations;
+        // Because we can have a huge amount of variant for 1 product, we also chunk update operations
+        await this.executeBulkOperationsByChunks(
+            this.options.reindexBulkOperationSizeLimit,
+            operations,
+            index,
+        )
+        return
     }
 
     private async getProductIdsByVariantIds(variantIds: ID[]): Promise<ID[]> {

packages/elasticsearch-plugin/src/options.ts

@@ -712,7 +712,7 @@ export const defaultOptions: ElasticsearchRuntimeOptions = {
     indexPrefix: 'vendure-',
     indexSettings: {},
     indexMappingProperties: {},
-    reindexProductsChunkSize: 500,
+    reindexProductsChunkSize: 2500,
     reindexBulkOperationSizeLimit: 3000,
     searchConfig: {
         facetValueMaxSize: 50,