fix(elasticsearch-plugin): Compatible with UUID primary keys strategy

Fixes #494

packages/common/src/generated-shop-types.ts

@@ -456,6 +456,8 @@ export type SearchInput = {
     take?: Maybe<Scalars['Int']>;
     skip?: Maybe<Scalars['Int']>;
     sort?: Maybe<SearchResultSortParameter>;
+    priceRange?: Maybe<PriceRangeInput>;
+    priceRangeWithTax?: Maybe<PriceRangeInput>;
 };
 
 export type SearchResultSortParameter = {
@@ -2140,6 +2142,7 @@ export type SearchResponse = {
     items: Array<SearchResult>;
     totalItems: Scalars['Int'];
     facetValues: Array<FacetValueResult>;
+    prices: SearchResponsePriceData;
 };
 
 /**
@@ -2454,6 +2457,25 @@ export type Zone = Node & {
     members: Array<Country>;
 };
 
+export type SearchResponsePriceData = {
+    __typename?: 'SearchResponsePriceData';
+    range: PriceRange;
+    rangeWithTax: PriceRange;
+    buckets: Array<PriceRangeBucket>;
+    bucketsWithTax: Array<PriceRangeBucket>;
+};
+
+export type PriceRangeBucket = {
+    __typename?: 'PriceRangeBucket';
+    to: Scalars['Int'];
+    count: Scalars['Int'];
+};
+
+export type PriceRangeInput = {
+    min: Scalars['Int'];
+    max: Scalars['Int'];
+};
+
 export type CollectionListOptions = {
     skip?: Maybe<Scalars['Int']>;
     take?: Maybe<Scalars['Int']>;

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

@@ -448,6 +448,8 @@ export type SearchInput = {
     take?: Maybe<Scalars['Int']>;
     skip?: Maybe<Scalars['Int']>;
     sort?: Maybe<SearchResultSortParameter>;
+    priceRange?: Maybe<PriceRangeInput>;
+    priceRangeWithTax?: Maybe<PriceRangeInput>;
 };
 
 export type SearchResultSortParameter = {
@@ -2054,6 +2056,7 @@ export type SearchResponse = {
     items: Array<SearchResult>;
     totalItems: Scalars['Int'];
     facetValues: Array<FacetValueResult>;
+    prices: SearchResponsePriceData;
 };
 
 /**
@@ -2341,6 +2344,23 @@ export type Zone = Node & {
     members: Array<Country>;
 };
 
+export type SearchResponsePriceData = {
+    range: PriceRange;
+    rangeWithTax: PriceRange;
+    buckets: Array<PriceRangeBucket>;
+    bucketsWithTax: Array<PriceRangeBucket>;
+};
+
+export type PriceRangeBucket = {
+    to: Scalars['Int'];
+    count: Scalars['Int'];
+};
+
+export type PriceRangeInput = {
+    min: Scalars['Int'];
+    max: Scalars['Int'];
+};
+
 export type CollectionListOptions = {
     skip?: Maybe<Scalars['Int']>;
     take?: Maybe<Scalars['Int']>;

packages/dev-server/dev-config.ts

@@ -63,10 +63,10 @@ export const devConfig: VendureConfig = {
         }),
         DefaultSearchPlugin,
         DefaultJobQueuePlugin,
-        // ElasticsearchPlugin.init({
-        //     host: 'http://localhost',
-        //     port: 9200,
-        // }),
+        /*ElasticsearchPlugin.init({
+            host: 'http://localhost',
+            port: 9200,
+        }),*/
         EmailPlugin.init({
             devMode: true,
             handlers: defaultEmailHandlers,

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

@@ -0,0 +1,127 @@
+import { DefaultJobQueuePlugin, DefaultLogger, LogLevel, mergeConfig, UuidIdStrategy } from '@vendure/core';
+import { createTestEnvironment } from '@vendure/testing';
+import gql from 'graphql-tag';
+import path from 'path';
+
+import { initialData } from '../../../e2e-common/e2e-initial-data';
+import { testConfig, TEST_SETUP_TIMEOUT_MS } from '../../../e2e-common/test-config';
+import { SearchProductsShop } from '../../core/e2e/graphql/generated-e2e-shop-types';
+import { SEARCH_PRODUCTS_SHOP } from '../../core/e2e/graphql/shop-definitions';
+import { awaitRunningJobs } from '../../core/e2e/utils/await-running-jobs';
+import { ElasticsearchPlugin } from '../src/plugin';
+
+import { GetCollectionList } from './graphql/generated-e2e-elasticsearch-plugin-types';
+// tslint:disable-next-line:no-var-requires
+const { elasticsearchHost, elasticsearchPort } = require('./constants');
+
+/**
+ * The Elasticsearch tests sometimes take a long time in CI due to limited resources.
+ * We increase the timeout to 30 seconds to prevent failure due to timeouts.
+ */
+if (process.env.CI) {
+    jest.setTimeout(10 * 3000);
+}
+
+// https://github.com/vendure-ecommerce/vendure/issues/494
+describe('Elasticsearch plugin with UuidIdStrategy', () => {
+    const { server, adminClient, shopClient } = createTestEnvironment(
+        mergeConfig(testConfig, {
+            apiOptions: {
+                port: 4050,
+            },
+            workerOptions: {
+                options: {
+                    port: 4055,
+                },
+            },
+            entityIdStrategy: new UuidIdStrategy(),
+            logger: new DefaultLogger({ level: LogLevel.Info }),
+            plugins: [
+                ElasticsearchPlugin.init({
+                    indexPrefix: 'e2e-uuid-tests',
+                    port: elasticsearchPort,
+                    host: elasticsearchHost,
+                }),
+                DefaultJobQueuePlugin,
+            ],
+        }),
+    );
+
+    beforeAll(async () => {
+        await server.init({
+            initialData,
+            productsCsvPath: path.join(__dirname, 'fixtures/e2e-products-full.csv'),
+            customerCount: 1,
+        });
+        await adminClient.asSuperAdmin();
+        await adminClient.query(REINDEX);
+        await awaitRunningJobs(adminClient);
+    }, TEST_SETUP_TIMEOUT_MS);
+
+    afterAll(async () => {
+        await server.destroy();
+    });
+
+    it('no term or filters', async () => {
+        const { search } = await shopClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
+            SEARCH_PRODUCTS_SHOP,
+            {
+                input: {
+                    groupByProduct: true,
+                },
+            },
+        );
+        expect(search.totalItems).toBe(20);
+    });
+
+    it('with search term', async () => {
+        const { search } = await shopClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
+            SEARCH_PRODUCTS_SHOP,
+            {
+                input: {
+                    groupByProduct: true,
+                    term: 'laptop',
+                },
+            },
+        );
+        expect(search.totalItems).toBe(1);
+    });
+
+    it('with collectionId filter term', async () => {
+        const { collections } = await shopClient.query<GetCollectionList.Query>(GET_COLLECTION_LIST);
+        const { search } = await shopClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
+            SEARCH_PRODUCTS_SHOP,
+            {
+                input: {
+                    groupByProduct: true,
+                    collectionId: collections.items[0].id,
+                },
+            },
+        );
+        expect(search.items).not.toEqual([]);
+    });
+});
+
+const REINDEX = gql`
+    mutation Reindex {
+        reindex {
+            id
+            queueName
+            state
+            progress
+            duration
+            result
+        }
+    }
+`;
+
+const GET_COLLECTION_LIST = gql`
+    query GetCollectionList {
+        collections {
+            items {
+                id
+                name
+            }
+        }
+    }
+`;

packages/elasticsearch-plugin/e2e/graphql/generated-e2e-elasticsearch-plugin-types.ts

@@ -3849,6 +3849,16 @@ export type NativeAuthInput = {
     password: Scalars['String'];
 };
 
+export type ReindexMutationVariables = Exact<{ [key: string]: never }>;
+
+export type ReindexMutation = {
+    reindex: Pick<Job, 'id' | 'queueName' | 'state' | 'progress' | 'duration' | 'result'>;
+};
+
+export type GetCollectionListQueryVariables = Exact<{ [key: string]: never }>;
+
+export type GetCollectionListQuery = { collections: { items: Array<Pick<Collection, 'id' | 'name'>> } };
+
 export type SearchProductsAdminQueryVariables = Exact<{
     input: SearchInput;
 }>;
@@ -3907,12 +3917,6 @@ export type SearchGetPricesQuery = {
     };
 };
 
-export type ReindexMutationVariables = Exact<{ [key: string]: never }>;
-
-export type ReindexMutation = {
-    reindex: Pick<Job, 'id' | 'queueName' | 'state' | 'progress' | 'duration' | 'result'>;
-};
-
 export type GetJobInfoQueryVariables = Exact<{
     id: Scalars['ID'];
 }>;
@@ -3923,6 +3927,21 @@ export type GetJobInfoQuery = {
 
 type DiscriminateUnion<T, U> = T extends U ? T : never;
 
+export namespace Reindex {
+    export type Variables = ReindexMutationVariables;
+    export type Mutation = ReindexMutation;
+    export type Reindex = NonNullable<ReindexMutation['reindex']>;
+}
+
+export namespace GetCollectionList {
+    export type Variables = GetCollectionListQueryVariables;
+    export type Query = GetCollectionListQuery;
+    export type Collections = NonNullable<GetCollectionListQuery['collections']>;
+    export type Items = NonNullable<
+        NonNullable<NonNullable<GetCollectionListQuery['collections']>['items']>[number]
+    >;
+}
+
 export namespace SearchProductsAdmin {
     export type Variables = SearchProductsAdminQueryVariables;
     export type Query = SearchProductsAdminQuery;
@@ -4013,12 +4032,6 @@ export namespace SearchGetPrices {
     >;
 }
 
-export namespace Reindex {
-    export type Variables = ReindexMutationVariables;
-    export type Mutation = ReindexMutation;
-    export type Reindex = NonNullable<ReindexMutation['reindex']>;
-}
-
 export namespace GetJobInfo {
     export type Variables = GetJobInfoQueryVariables;
     export type Query = GetJobInfoQuery;

packages/elasticsearch-plugin/src/constants.ts

@@ -1,6 +1,4 @@
 export const ELASTIC_SEARCH_OPTIONS = Symbol('ELASTIC_SEARCH_OPTIONS');
 export const VARIANT_INDEX_NAME = 'variants';
-export const VARIANT_INDEX_TYPE = 'variant-index-item';
 export const PRODUCT_INDEX_NAME = 'products';
-export const PRODUCT_INDEX_TYPE = 'product-index-item';
 export const loggerCtx = 'ElasticsearchPlugin';

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

@@ -2,6 +2,7 @@ import { Client, ClientOptions } from '@elastic/elasticsearch';
 import { Inject, Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
 import { SearchResult, SearchResultAsset } from '@vendure/common/lib/generated-types';
 import {
+    ConfigService,
     DeepRequired,
     FacetValue,
     FacetValueService,
@@ -13,14 +14,7 @@ import {
 } from '@vendure/core';
 
 import { buildElasticBody } from './build-elastic-body';
-import {
-    ELASTIC_SEARCH_OPTIONS,
-    loggerCtx,
-    PRODUCT_INDEX_NAME,
-    PRODUCT_INDEX_TYPE,
-    VARIANT_INDEX_NAME,
-    VARIANT_INDEX_TYPE,
-} from './constants';
+import { ELASTIC_SEARCH_OPTIONS, loggerCtx, PRODUCT_INDEX_NAME, VARIANT_INDEX_NAME } from './constants';
 import { ElasticsearchIndexService } from './elasticsearch-index.service';
 import { createIndices } from './indexing-utils';
 import { ElasticsearchOptions } from './options';
@@ -44,6 +38,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
         private searchService: SearchService,
         private elasticsearchIndexService: ElasticsearchIndexService,
         private facetValueService: FacetValueService,
+        private configService: ConfigService,
     ) {
         searchService.adopt(this);
     }
@@ -76,7 +71,11 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
 
             if (result.body === false) {
                 Logger.verbose(`Index "${index}" does not exist. Creating...`, loggerCtx);
-                await createIndices(this.client, indexPrefix);
+                await createIndices(
+                    this.client,
+                    indexPrefix,
+                    this.configService.entityIdStrategy.primaryKeyType,
+                );
             } else {
                 Logger.verbose(`Index "${index}" exists`, loggerCtx);
             }
@@ -104,25 +103,33 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
             enabledOnly,
         );
         if (groupByProduct) {
-            const { body }: { body: SearchResponseBody<ProductIndexItem> } = await this.client.search({
-                index: indexPrefix + PRODUCT_INDEX_NAME,
-                type: PRODUCT_INDEX_TYPE,
-                body: elasticSearchBody,
-            });
-            return {
-                items: body.hits.hits.map(hit => this.mapProductToSearchResult(hit)),
-                totalItems: body.hits.total.value,
-            };
+            try {
+                const { body }: { body: SearchResponseBody<ProductIndexItem> } = await this.client.search({
+                    index: indexPrefix + PRODUCT_INDEX_NAME,
+                    body: elasticSearchBody,
+                });
+                return {
+                    items: body.hits.hits.map(hit => this.mapProductToSearchResult(hit)),
+                    totalItems: body.hits.total.value,
+                };
+            } catch (e) {
+                Logger.error(e.message, loggerCtx, e.stack);
+                throw e;
+            }
         } else {
-            const { body }: { body: SearchResponseBody<VariantIndexItem> } = await this.client.search({
-                index: indexPrefix + VARIANT_INDEX_NAME,
-                type: VARIANT_INDEX_TYPE,
-                body: elasticSearchBody,
-            });
-            return {
-                items: body.hits.hits.map(hit => this.mapVariantToSearchResult(hit)),
-                totalItems: body.hits.total.value,
-            };
+            try {
+                const { body }: { body: SearchResponseBody<VariantIndexItem> } = await this.client.search({
+                    index: indexPrefix + VARIANT_INDEX_NAME,
+                    body: elasticSearchBody,
+                });
+                return {
+                    items: body.hits.hits.map(hit => this.mapVariantToSearchResult(hit)),
+                    totalItems: body.hits.total.value,
+                };
+            } catch (e) {
+                Logger.error(e.message, loggerCtx, e.stack);
+                throw e;
+            }
         }
     }
 
@@ -147,16 +154,22 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
         elasticSearchBody.aggs = {
             facetValue: {
                 terms: {
-                    field: 'facetValueIds.keyword',
+                    field: 'facetValueIds',
                     size: this.options.searchConfig.facetValueMaxSize,
                 },
             },
         };
-        const { body }: { body: SearchResponseBody<VariantIndexItem> } = await this.client.search({
-            index: indexPrefix + (input.groupByProduct ? PRODUCT_INDEX_NAME : VARIANT_INDEX_NAME),
-            type: input.groupByProduct ? PRODUCT_INDEX_TYPE : VARIANT_INDEX_TYPE,
-            body: elasticSearchBody,
-        });
+        let body: SearchResponseBody<VariantIndexItem>;
+        try {
+            const result = await this.client.search<SearchResponseBody<VariantIndexItem>>({
+                index: indexPrefix + (input.groupByProduct ? PRODUCT_INDEX_NAME : VARIANT_INDEX_NAME),
+                body: elasticSearchBody,
+            });
+            body = result.body;
+        } catch (e) {
+            Logger.error(e.message, loggerCtx, e.stack);
+            throw e;
+        }
 
         const buckets = body.aggregations ? body.aggregations.facetValue.buckets : [];
 
@@ -221,7 +234,6 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
         };
         const { body }: { body: SearchResponseBody<VariantIndexItem> } = await this.client.search({
             index: indexPrefix + (input.groupByProduct ? PRODUCT_INDEX_NAME : VARIANT_INDEX_NAME),
-            type: input.groupByProduct ? PRODUCT_INDEX_TYPE : VARIANT_INDEX_TYPE,
             body: elasticSearchBody,
         });
 

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

@@ -24,14 +24,7 @@ import { Observable } from 'rxjs';
 import { SelectQueryBuilder } from 'typeorm';
 import { FindOptionsUtils } from 'typeorm/find-options/FindOptionsUtils';
 
-import {
-    ELASTIC_SEARCH_OPTIONS,
-    loggerCtx,
-    PRODUCT_INDEX_NAME,
-    PRODUCT_INDEX_TYPE,
-    VARIANT_INDEX_NAME,
-    VARIANT_INDEX_TYPE,
-} from './constants';
+import { ELASTIC_SEARCH_OPTIONS, loggerCtx, PRODUCT_INDEX_NAME, VARIANT_INDEX_NAME } from './constants';
 import { createIndices, deleteByChannel, deleteIndices } from './indexing-utils';
 import { ElasticsearchOptions } from './options';
 import {
@@ -305,7 +298,11 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
 
                 if (dropIndices) {
                     await deleteIndices(this.client, this.options.indexPrefix);
-                    await createIndices(this.client, this.options.indexPrefix);
+                    await createIndices(
+                        this.client,
+                        this.options.indexPrefix,
+                        this.configService.entityIdStrategy.primaryKeyType,
+                    );
                 } else {
                     await deleteByChannel(this.client, this.options.indexPrefix, ctx.channelId);
                 }
@@ -486,8 +483,8 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                 productIdsIndexed.add(variant.productId);
             }
         }
-        await this.executeBulkOperations(VARIANT_INDEX_NAME, VARIANT_INDEX_TYPE, variantsToIndex);
-        await this.executeBulkOperations(PRODUCT_INDEX_NAME, PRODUCT_INDEX_TYPE, productsToIndex);
+        await this.executeBulkOperations(VARIANT_INDEX_NAME, variantsToIndex);
+        await this.executeBulkOperations(PRODUCT_INDEX_NAME, productsToIndex);
         return variantsInProduct;
     }
 
@@ -526,7 +523,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                 }
             }
             Logger.verbose(`Updating ${updatedVariants.length} ProductVariants`, loggerCtx);
-            await this.executeBulkOperations(VARIANT_INDEX_NAME, VARIANT_INDEX_TYPE, operations);
+            await this.executeBulkOperations(VARIANT_INDEX_NAME, operations);
         }
     }
 
@@ -569,7 +566,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                         { doc: updatedProductIndexItem, doc_as_upsert: true },
                     );
                 }
-                await this.executeBulkOperations(PRODUCT_INDEX_NAME, PRODUCT_INDEX_TYPE, operations);
+                await this.executeBulkOperations(PRODUCT_INDEX_NAME, operations);
             }
         }
     }
@@ -581,7 +578,7 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
         for (const languageCode of languageVariants) {
             operations.push({ delete: { _id: this.getId(product.id, channelId, languageCode) } });
         }
-        await this.executeBulkOperations(PRODUCT_INDEX_NAME, PRODUCT_INDEX_TYPE, operations);
+        await this.executeBulkOperations(PRODUCT_INDEX_NAME, operations);
     }
 
     private async deleteVariantsInternal(variants: ProductVariant[], channelId: ID) {
@@ -595,12 +592,11 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
                 });
             }
         }
-        await this.executeBulkOperations(VARIANT_INDEX_NAME, VARIANT_INDEX_TYPE, operations);
+        await this.executeBulkOperations(VARIANT_INDEX_NAME, operations);
     }
 
     private async executeBulkOperations(
         indexName: string,
-        indexType: string,
         operations: Array<BulkOperation | BulkOperationDoc<VariantIndexItem | ProductIndexItem>>,
     ) {
         try {
@@ -608,13 +604,12 @@ export class ElasticsearchIndexerController implements OnModuleInit, OnModuleDes
             const { body }: { body: BulkResponseBody } = await this.client.bulk({
                 refresh: true,
                 index: fullIndexName,
-                type: indexType,
                 body: operations,
             });
 
             if (body.errors) {
                 Logger.error(
-                    `Some errors occurred running bulk operations on ${indexType}! Set logger to "debug" to print all errors.`,
+                    `Some errors occurred running bulk operations on ${fullIndexName}! Set logger to "debug" to print all errors.`,
                     loggerCtx,
                 );
                 body.items.forEach(item => {

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

@@ -2,18 +2,82 @@ import { Client } from '@elastic/elasticsearch';
 import { ID, Logger } from '@vendure/core';
 
 import { loggerCtx, PRODUCT_INDEX_NAME, VARIANT_INDEX_NAME } from './constants';
+import { ProductIndexItem, VariantIndexItem } from './types';
+
+export async function createIndices(client: Client, prefix: string, primaryKeyType: 'increment' | 'uuid') {
+    const textWithKeyword = {
+        type: 'text',
+        fields: {
+            keyword: {
+                type: 'keyword',
+                ignore_above: 256,
+            },
+        },
+    };
+    const keyword = { type: 'keyword' };
+    const commonMappings = {
+        sku: textWithKeyword,
+        slug: textWithKeyword,
+        productId: keyword,
+        channelId: keyword,
+        languageCode: keyword,
+        productName: textWithKeyword,
+        productVariantId: keyword,
+        productVariantName: textWithKeyword,
+        currencyCode: keyword,
+        description: textWithKeyword,
+        facetIds: keyword,
+        facetValueIds: keyword,
+        collectionIds: keyword,
+        collectionSlugs: keyword,
+        channelIds: keyword,
+        enabled: { type: 'boolean' },
+        productAssetId: keyword,
+        productPreview: textWithKeyword,
+        productPreviewFocalPoint: { type: 'object' },
+        productVariantAssetId: keyword,
+        productVariantPreview: textWithKeyword,
+        productVariantPreviewFocalPoint: { type: 'object' },
+    };
+
+    const productMappings: { [prop in keyof ProductIndexItem]: any } = {
+        ...commonMappings,
+        priceMin: { type: 'long' },
+        priceMax: { type: 'long' },
+        priceWithTaxMin: { type: 'long' },
+        priceWithTaxMax: { type: 'long' },
+    };
+
+    const variantMappings: { [prop in keyof VariantIndexItem]: any } = {
+        ...commonMappings,
+        price: { type: 'long' },
+        priceWithTax: { type: 'long' },
+    };
 
-export async function createIndices(client: Client, prefix: string) {
     try {
         const index = prefix + VARIANT_INDEX_NAME;
-        await client.indices.create({ index });
+        await client.indices.create({
+            index,
+            body: {
+                mappings: {
+                    properties: variantMappings,
+                },
+            },
+        });
         Logger.verbose(`Created index "${index}"`, loggerCtx);
     } catch (e) {
         Logger.error(JSON.stringify(e, null, 2), loggerCtx);
     }
     try {
         const index = prefix + PRODUCT_INDEX_NAME;
-        await client.indices.create({ index });
+        await client.indices.create({
+            index,
+            body: {
+                mappings: {
+                    properties: productMappings,
+                },
+            },
+        });
         Logger.verbose(`Created index "${index}"`, loggerCtx);
     } catch (e) {
         Logger.error(JSON.stringify(e, null, 2), loggerCtx);