feat(core): Make product/variant events more granular

BREAKING CHANGE: The `CatalogModificationEvent` which was previously published whenever changes were made to `Product` or `ProductVariant` entities has been replaced with a `ProductEvent` and `ProductVariantEvent`, including the type of event ('created', 'updated', 'deleted').

packages/core/e2e/default-search-plugin.e2e-spec.ts

@@ -1,6 +1,5 @@
 import { pick } from '@vendure/common/lib/pick';
-import { mergeConfig } from '@vendure/core';
-import { DefaultSearchPlugin } from '@vendure/core';
+import { DefaultSearchPlugin, mergeConfig } from '@vendure/core';
 import { facetValueCollectionFilter } from '@vendure/core/dist/config/collection/default-collection-filters';
 import { createTestEnvironment, SimpleGraphQLClient } from '@vendure/testing';
 import gql from 'graphql-tag';
@@ -11,6 +10,8 @@ import { initialData } from './fixtures/e2e-initial-data';
 import {
     CreateCollection,
     CreateFacet,
+    DeleteProduct,
+    DeleteProductVariant,
     LanguageCode,
     SearchFacetValues,
     SearchGetPrices,
@@ -24,6 +25,8 @@ import { SearchProductsShop } from './graphql/generated-e2e-shop-types';
 import {
     CREATE_COLLECTION,
     CREATE_FACET,
+    DELETE_PRODUCT,
+    DELETE_PRODUCT_VARIANT,
     UPDATE_COLLECTION,
     UPDATE_PRODUCT,
     UPDATE_PRODUCT_VARIANTS,
@@ -51,6 +54,12 @@ describe('Default search plugin', () => {
         await server.destroy();
     });
 
+    function doAdminSearchQuery(input: SearchInput) {
+        return adminClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(SEARCH_PRODUCTS, {
+            input,
+        });
+    }
+
     async function testGroupByProduct(client: SimpleGraphQLClient) {
         const result = await client.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
             SEARCH_PRODUCTS_SHOP,
@@ -344,45 +353,160 @@ describe('Default search plugin', () => {
 
         it('price ranges', () => testPriceRanges(adminClient));
 
-        it('updates index when a Product is changed', async () => {
-            await adminClient.query<UpdateProduct.Mutation, UpdateProduct.Variables>(UPDATE_PRODUCT, {
-                input: {
-                    id: 'T_1',
-                    facetValueIds: [],
-                },
+        describe('updating the index', () => {
+            it('updates index when ProductVariants are changed', async () => {
+                await awaitRunningJobs(adminClient);
+                const { search } = await doAdminSearchQuery({ term: 'drive', groupByProduct: false });
+                expect(search.items.map(i => i.sku)).toEqual([
+                    'IHD455T1',
+                    'IHD455T2',
+                    'IHD455T3',
+                    'IHD455T4',
+                    'IHD455T6',
+                ]);
+
+                await adminClient.query<UpdateProductVariants.Mutation, UpdateProductVariants.Variables>(
+                    UPDATE_PRODUCT_VARIANTS,
+                    {
+                        input: search.items.map(i => ({
+                            id: i.productVariantId,
+                            sku: i.sku + '_updated',
+                        })),
+                    },
+                );
+
+                await awaitRunningJobs(adminClient);
+                const { search: search2 } = await doAdminSearchQuery({
+                    term: 'drive',
+                    groupByProduct: false,
+                });
+
+                expect(search2.items.map(i => i.sku)).toEqual([
+                    'IHD455T1_updated',
+                    'IHD455T2_updated',
+                    'IHD455T3_updated',
+                    'IHD455T4_updated',
+                    'IHD455T6_updated',
+                ]);
             });
-            await awaitRunningJobs(adminClient);
-            const result = await adminClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
-                SEARCH_PRODUCTS,
-                {
+
+            it('updates index when ProductVariants are deleted', async () => {
+                await awaitRunningJobs(adminClient);
+                const { search } = await doAdminSearchQuery({ term: 'drive', groupByProduct: false });
+
+                await adminClient.query<DeleteProductVariant.Mutation, DeleteProductVariant.Variables>(
+                    DELETE_PRODUCT_VARIANT,
+                    {
+                        id: search.items[0].productVariantId,
+                    },
+                );
+
+                await awaitRunningJobs(adminClient);
+                const { search: search2 } = await doAdminSearchQuery({
+                    term: 'drive',
+                    groupByProduct: false,
+                });
+
+                expect(search2.items.map(i => i.sku)).toEqual([
+                    'IHD455T2_updated',
+                    'IHD455T3_updated',
+                    'IHD455T4_updated',
+                    'IHD455T6_updated',
+                ]);
+            });
+
+            it('updates index when a Product is changed', async () => {
+                await adminClient.query<UpdateProduct.Mutation, UpdateProduct.Variables>(UPDATE_PRODUCT, {
                     input: {
-                        facetValueIds: ['T_2'],
-                        groupByProduct: true,
+                        id: 'T_1',
+                        facetValueIds: [],
                     },
-                },
-            );
-            expect(result.search.items.map(i => i.productName)).toEqual([
-                'Curvy Monitor',
-                'Gaming PC',
-                'Hard Drive',
-                'Clacky Keyboard',
-                'USB Cable',
-            ]);
-        });
+                });
+                await awaitRunningJobs(adminClient);
+                const result = await doAdminSearchQuery({ facetValueIds: ['T_2'], groupByProduct: true });
+                expect(result.search.items.map(i => i.productName)).toEqual([
+                    'Curvy Monitor',
+                    'Gaming PC',
+                    'Hard Drive',
+                    'Clacky Keyboard',
+                    'USB Cable',
+                ]);
+            });
 
-        it('updates index when a Collection is changed', async () => {
-            await adminClient.query<UpdateCollection.Mutation, UpdateCollection.Variables>(
-                UPDATE_COLLECTION,
-                {
+            it('updates index when a Product is deleted', async () => {
+                const { search } = await doAdminSearchQuery({ facetValueIds: ['T_2'], groupByProduct: true });
+                expect(search.items.map(i => i.productId)).toEqual(['T_2', 'T_3', 'T_4', 'T_5', 'T_6']);
+                await adminClient.query<DeleteProduct.Mutation, DeleteProduct.Variables>(DELETE_PRODUCT, {
+                    id: 'T_5',
+                });
+                await awaitRunningJobs(adminClient);
+                const { search: search2 } = await doAdminSearchQuery({
+                    facetValueIds: ['T_2'],
+                    groupByProduct: true,
+                });
+                expect(search2.items.map(i => i.productId)).toEqual(['T_2', 'T_3', 'T_4', 'T_6']);
+            });
+
+            it('updates index when a Collection is changed', async () => {
+                await adminClient.query<UpdateCollection.Mutation, UpdateCollection.Variables>(
+                    UPDATE_COLLECTION,
+                    {
+                        input: {
+                            id: 'T_2',
+                            filters: [
+                                {
+                                    code: facetValueCollectionFilter.code,
+                                    arguments: [
+                                        {
+                                            name: 'facetValueIds',
+                                            value: `["T_4"]`,
+                                            type: 'facetValueIds',
+                                        },
+                                        {
+                                            name: 'containsAny',
+                                            value: `false`,
+                                            type: 'boolean',
+                                        },
+                                    ],
+                                },
+                            ],
+                        },
+                    },
+                );
+                await awaitRunningJobs(adminClient);
+                const result = await doAdminSearchQuery({ collectionId: 'T_2', groupByProduct: true });
+
+                expect(result.search.items.map(i => i.productName)).toEqual([
+                    'Road Bike',
+                    'Skipping Rope',
+                    'Boxing Gloves',
+                    'Tent',
+                    'Cruiser Skateboard',
+                    'Football',
+                    'Running Shoe',
+                ]);
+            });
+
+            it('updates index when a Collection created', async () => {
+                const { createCollection } = await adminClient.query<
+                    CreateCollection.Mutation,
+                    CreateCollection.Variables
+                >(CREATE_COLLECTION, {
                     input: {
-                        id: 'T_2',
+                        translations: [
+                            {
+                                languageCode: LanguageCode.en,
+                                name: 'Photo',
+                                description: '',
+                            },
+                        ],
                         filters: [
                             {
                                 code: facetValueCollectionFilter.code,
                                 arguments: [
                                     {
                                         name: 'facetValueIds',
-                                        value: `["T_4"]`,
+                                        value: `["T_3"]`,
                                         type: 'facetValueIds',
                                     },
                                     {
@@ -394,193 +518,103 @@ describe('Default search plugin', () => {
                             },
                         ],
                     },
-                },
-            );
-            await awaitRunningJobs(adminClient);
-            const result = await adminClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
-                SEARCH_PRODUCTS,
-                {
-                    input: {
-                        collectionId: 'T_2',
-                        groupByProduct: true,
-                    },
-                },
-            );
-            expect(result.search.items.map(i => i.productName)).toEqual([
-                'Road Bike',
-                'Skipping Rope',
-                'Boxing Gloves',
-                'Tent',
-                'Cruiser Skateboard',
-                'Football',
-                'Running Shoe',
-            ]);
-        });
-
-        it('updates index when a Collection created', async () => {
-            const { createCollection } = await adminClient.query<
-                CreateCollection.Mutation,
-                CreateCollection.Variables
-            >(CREATE_COLLECTION, {
-                input: {
-                    translations: [
-                        {
-                            languageCode: LanguageCode.en,
-                            name: 'Photo',
-                            description: '',
-                        },
-                    ],
-                    filters: [
-                        {
-                            code: facetValueCollectionFilter.code,
-                            arguments: [
-                                {
-                                    name: 'facetValueIds',
-                                    value: `["T_3"]`,
-                                    type: 'facetValueIds',
-                                },
-                                {
-                                    name: 'containsAny',
-                                    value: `false`,
-                                    type: 'boolean',
-                                },
-                            ],
-                        },
-                    ],
-                },
+                });
+                await awaitRunningJobs(adminClient);
+                const result = await doAdminSearchQuery({
+                    collectionId: createCollection.id,
+                    groupByProduct: true,
+                });
+                expect(result.search.items.map(i => i.productName)).toEqual([
+                    'Instant Camera',
+                    'Camera Lens',
+                    'Tripod',
+                    'SLR Camera',
+                ]);
             });
-            await awaitRunningJobs(adminClient);
-            const result = await adminClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
-                SEARCH_PRODUCTS,
-                {
+
+            it('updates index when a taxRate is changed', async () => {
+                await adminClient.query<UpdateTaxRate.Mutation, UpdateTaxRate.Variables>(UPDATE_TAX_RATE, {
                     input: {
-                        collectionId: createCollection.id,
-                        groupByProduct: true,
+                        // Default Channel's defaultTaxZone is Europe (id 2) and the id of the standard TaxRate
+                        // to Europe is 2.
+                        id: 'T_2',
+                        value: 50,
                     },
-                },
-            );
-            expect(result.search.items.map(i => i.productName)).toEqual([
-                'Instant Camera',
-                'Camera Lens',
-                'Tripod',
-                'SLR Camera',
-            ]);
-        });
+                });
+                await awaitRunningJobs(adminClient);
+                const result = await adminClient.query<SearchGetPrices.Query, SearchGetPrices.Variables>(
+                    SEARCH_GET_PRICES,
+                    {
+                        input: {
+                            groupByProduct: true,
+                            term: 'laptop',
+                        } as SearchInput,
+                    },
+                );
+                expect(result.search.items).toEqual([
+                    {
+                        price: { min: 129900, max: 229900 },
+                        priceWithTax: { min: 194850, max: 344850 },
+                    },
+                ]);
+            });
 
-        it('updates index when a taxRate is changed', async () => {
-            await adminClient.query<UpdateTaxRate.Mutation, UpdateTaxRate.Variables>(UPDATE_TAX_RATE, {
-                input: {
-                    // Default Channel's defaultTaxZone is Europe (id 2) and the id of the standard TaxRate
-                    // to Europe is 2.
-                    id: 'T_2',
-                    value: 50,
-                },
+            it('returns disabled field when not grouped', async () => {
+                const result = await doAdminSearchQuery({ groupByProduct: false, take: 3 });
+                expect(result.search.items.map(pick(['productVariantId', 'enabled']))).toEqual([
+                    { productVariantId: 'T_1', enabled: true },
+                    { productVariantId: 'T_2', enabled: true },
+                    { productVariantId: 'T_3', enabled: false },
+                ]);
             });
-            await awaitRunningJobs(adminClient);
-            const result = await adminClient.query<SearchGetPrices.Query, SearchGetPrices.Variables>(
-                SEARCH_GET_PRICES,
-                {
-                    input: {
-                        groupByProduct: true,
-                        term: 'laptop',
-                    } as SearchInput,
-                },
-            );
-            expect(result.search.items).toEqual([
-                {
-                    price: { min: 129900, max: 229900 },
-                    priceWithTax: { min: 194850, max: 344850 },
-                },
-            ]);
-        });
 
-        it('returns disabled field when not grouped', async () => {
-            const result = await adminClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
-                SEARCH_PRODUCTS,
-                {
-                    input: {
-                        groupByProduct: false,
-                        take: 3,
+            it('when grouped, disabled is false if at least one variant is enabled', async () => {
+                await adminClient.query<UpdateProductVariants.Mutation, UpdateProductVariants.Variables>(
+                    UPDATE_PRODUCT_VARIANTS,
+                    {
+                        input: [{ id: 'T_1', enabled: false }, { id: 'T_2', enabled: false }],
                     },
-                },
-            );
-            expect(result.search.items.map(pick(['productVariantId', 'enabled']))).toEqual([
-                { productVariantId: 'T_1', enabled: true },
-                { productVariantId: 'T_2', enabled: true },
-                { productVariantId: 'T_3', enabled: false },
-            ]);
-        });
+                );
+                await awaitRunningJobs(adminClient);
+                const result = await doAdminSearchQuery({ groupByProduct: true, take: 3 });
+                expect(result.search.items.map(pick(['productId', 'enabled']))).toEqual([
+                    { productId: 'T_1', enabled: true },
+                    { productId: 'T_2', enabled: true },
+                    { productId: 'T_3', enabled: true },
+                ]);
+            });
 
-        it('when grouped, disabled is false if at least one variant is enabled', async () => {
-            await adminClient.query<UpdateProductVariants.Mutation, UpdateProductVariants.Variables>(
-                UPDATE_PRODUCT_VARIANTS,
-                {
-                    input: [{ id: 'T_1', enabled: false }, { id: 'T_2', enabled: false }],
-                },
-            );
-            await awaitRunningJobs(adminClient);
-            const result = await adminClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
-                SEARCH_PRODUCTS,
-                {
-                    input: {
-                        groupByProduct: true,
-                        take: 3,
+            it('when grouped, disabled is true if all variants are disabled', async () => {
+                await adminClient.query<UpdateProductVariants.Mutation, UpdateProductVariants.Variables>(
+                    UPDATE_PRODUCT_VARIANTS,
+                    {
+                        input: [{ id: 'T_4', enabled: false }],
                     },
-                },
-            );
-            expect(result.search.items.map(pick(['productId', 'enabled']))).toEqual([
-                { productId: 'T_1', enabled: true },
-                { productId: 'T_2', enabled: true },
-                { productId: 'T_3', enabled: true },
-            ]);
-        });
+                );
+                await awaitRunningJobs(adminClient);
+                const result = await doAdminSearchQuery({ groupByProduct: true, take: 3 });
+                expect(result.search.items.map(pick(['productId', 'enabled']))).toEqual([
+                    { productId: 'T_1', enabled: false },
+                    { productId: 'T_2', enabled: true },
+                    { productId: 'T_3', enabled: true },
+                ]);
+            });
 
-        it('when grouped, disabled is true if all variants are disabled', async () => {
-            await adminClient.query<UpdateProductVariants.Mutation, UpdateProductVariants.Variables>(
-                UPDATE_PRODUCT_VARIANTS,
-                {
-                    input: [{ id: 'T_4', enabled: false }],
-                },
-            );
-            await awaitRunningJobs(adminClient);
-            const result = await adminClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
-                SEARCH_PRODUCTS,
-                {
+            it('when grouped, disabled is true product is disabled', async () => {
+                await adminClient.query<UpdateProduct.Mutation, UpdateProduct.Variables>(UPDATE_PRODUCT, {
                     input: {
-                        groupByProduct: true,
-                        take: 3,
+                        id: 'T_3',
+                        enabled: false,
                     },
-                },
-            );
-            expect(result.search.items.map(pick(['productId', 'enabled']))).toEqual([
-                { productId: 'T_1', enabled: false },
-                { productId: 'T_2', enabled: true },
-                { productId: 'T_3', enabled: true },
-            ]);
-        });
-
-        it('when grouped, disabled is true product is disabled', async () => {
-            await adminClient.query<UpdateProduct.Mutation, UpdateProduct.Variables>(UPDATE_PRODUCT, {
-                input: {
-                    id: 'T_3',
-                    enabled: false,
-                },
+                });
+                await awaitRunningJobs(adminClient);
+                const result = await doAdminSearchQuery({ groupByProduct: true, take: 3 });
+                expect(result.search.items.map(pick(['productId', 'enabled']))).toEqual([
+                    { productId: 'T_1', enabled: false },
+                    { productId: 'T_2', enabled: true },
+                    { productId: 'T_3', enabled: false },
+                ]);
             });
-            await awaitRunningJobs(adminClient);
-            const result = await adminClient.query<SearchProductsShop.Query, SearchProductsShop.Variables>(
-                SEARCH_PRODUCTS,
-                {
-                    input: {
-                        groupByProduct: true,
-                        take: 3,
-                    },
-                },
-            );
-            expect(result.search.items.map(pick(['productId', 'enabled']))).toEqual([
-                { productId: 'T_1', enabled: false },
-                { productId: 'T_2', enabled: true },
-                { productId: 'T_3', enabled: false },
-            ]);
         });
     });
 });

packages/core/e2e/graphql/shared-definitions.ts

@@ -310,3 +310,12 @@ export const CREATE_CHANNEL = gql`
         }
     }
 `;
+
+export const DELETE_PRODUCT_VARIANT = gql`
+    mutation DeleteProductVariant($id: ID!) {
+        deleteProductVariant(id: $id) {
+            result
+            message
+        }
+    }
+`;

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

@@ -30,6 +30,7 @@ import {
     CREATE_PRODUCT,
     CREATE_PRODUCT_VARIANTS,
     DELETE_PRODUCT,
+    DELETE_PRODUCT_VARIANT,
     GET_ASSET_LIST,
     GET_PRODUCT_LIST,
     GET_PRODUCT_SIMPLE,
@@ -1125,12 +1126,3 @@ export const GET_OPTION_GROUP = gql`
         }
     }
 `;
-
-export const DELETE_PRODUCT_VARIANT = gql`
-    mutation DeleteProductVariant($id: ID!) {
-        deleteProductVariant(id: $id) {
-            result
-            message
-        }
-    }
-`;

packages/core/src/api/resolvers/admin/product.resolver.ts

@@ -120,7 +120,7 @@ export class ProductResolver {
         @Args() args: MutationCreateProductVariantsArgs,
     ): Promise<Array<Translated<ProductVariant>>> {
         const { input } = args;
-        return Promise.all(input.map(i => this.productVariantService.create(ctx, i)));
+        return this.productVariantService.create(ctx, input);
     }
 
     @Mutation()
@@ -130,7 +130,7 @@ export class ProductResolver {
         @Args() args: MutationUpdateProductVariantsArgs,
     ): Promise<Array<Translated<ProductVariant>>> {
         const { input } = args;
-        return Promise.all(input.map(i => this.productVariantService.update(ctx, i)));
+        return this.productVariantService.update(ctx, input);
     }
 
     @Mutation()

packages/core/src/api/schema/admin-api/product-search.api.graphql

@@ -8,4 +8,6 @@ type Mutation {
 
 type SearchResult {
     enabled: Boolean!
+    "An array of ids of the Collections in which this result appears"
+    channelIds: [ID!]!
 }

packages/core/src/event-bus/events/catalog-modification-event.ts

@@ -1,18 +0,0 @@
-import { RequestContext } from '../../api/common/request-context';
-import { Product, ProductVariant } from '../../entity';
-import { VendureEvent } from '../vendure-event';
-
-/**
- * @description
- * This event is fired whenever the catalog is modified in some way, i.e. a
- * {@link Product} or {@link ProductVariant} is modified is created, updated, or
- * deleted.
- *
- * @docsCategory events
- * @docsPage Event Types
- */
-export class CatalogModificationEvent extends VendureEvent {
-    constructor(public ctx: RequestContext, public entity: Product | ProductVariant) {
-        super();
-    }
-}

packages/core/src/event-bus/events/product-channel-event.ts

@@ -0,0 +1,24 @@
+import { ID } from '@vendure/common/lib/shared-types';
+
+import { RequestContext } from '../../api/common/request-context';
+import { Product } from '../../entity';
+import { VendureEvent } from '../vendure-event';
+
+/**
+ * @description
+ * This event is fired whenever a {@link Product} is added, updated
+ * or deleted.
+ *
+ * @docsCategory events
+ * @docsPage Event Types
+ */
+export class ProductChannelEvent extends VendureEvent {
+    constructor(
+        public ctx: RequestContext,
+        public product: Product,
+        public channelId: ID,
+        public type: 'assigned' | 'removed',
+    ) {
+        super();
+    }
+}

packages/core/src/event-bus/events/product-event.ts

@@ -0,0 +1,21 @@
+import { RequestContext } from '../../api/common/request-context';
+import { Product } from '../../entity';
+import { VendureEvent } from '../vendure-event';
+
+/**
+ * @description
+ * This event is fired whenever a {@link Product} is added, updated
+ * or deleted.
+ *
+ * @docsCategory events
+ * @docsPage Event Types
+ */
+export class ProductEvent extends VendureEvent {
+    constructor(
+        public ctx: RequestContext,
+        public product: Product,
+        public type: 'created' | 'updated' | 'deleted',
+    ) {
+        super();
+    }
+}

packages/core/src/event-bus/events/product-variant-event.ts

@@ -0,0 +1,21 @@
+import { RequestContext } from '../../api/common/request-context';
+import { ProductVariant } from '../../entity';
+import { VendureEvent } from '../vendure-event';
+
+/**
+ * @description
+ * This event is fired whenever a {@link ProductVariant} is added, updated
+ * or deleted.
+ *
+ * @docsCategory events
+ * @docsPage Event Types
+ */
+export class ProductVariantEvent extends VendureEvent {
+    constructor(
+        public ctx: RequestContext,
+        public variants: ProductVariant[],
+        public type: 'created' | 'updated' | 'deleted',
+    ) {
+        super();
+    }
+}

packages/core/src/event-bus/index.ts

@@ -2,7 +2,9 @@ export * from './event-bus';
 export * from './event-bus.module';
 export * from './vendure-event';
 export * from './events/account-registration-event';
-export * from './events/catalog-modification-event';
+export * from './events/product-event';
+export * from './events/product-channel-event';
+export * from './events/product-variant-event';
 export * from './events/collection-modification-event';
 export * from './events/identifier-change-event';
 export * from './events/identifier-change-request-event';

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

@@ -12,6 +12,7 @@ import {
 import { pick } from '@vendure/common/lib/pick';
 import { ROOT_COLLECTION_NAME } from '@vendure/common/lib/shared-constants';
 import { ID, PaginatedList } from '@vendure/common/lib/shared-types';
+import { merge } from 'rxjs';
 import { debounceTime } from 'rxjs/operators';
 import { Connection } from 'typeorm';
 
@@ -32,8 +33,9 @@ import { CollectionTranslation } from '../../entity/collection/collection-transl
 import { Collection } from '../../entity/collection/collection.entity';
 import { ProductVariant } from '../../entity/product-variant/product-variant.entity';
 import { EventBus } from '../../event-bus/event-bus';
-import { CatalogModificationEvent } from '../../event-bus/events/catalog-modification-event';
 import { CollectionModificationEvent } from '../../event-bus/events/collection-modification-event';
+import { ProductEvent } from '../../event-bus/events/product-event';
+import { ProductVariantEvent } from '../../event-bus/events/product-variant-event';
 import { WorkerService } from '../../worker/worker.service';
 import { ListQueryBuilder } from '../helpers/list-query-builder/list-query-builder';
 import { TranslatableSaver } from '../helpers/translatable-saver/translatable-saver';
@@ -68,8 +70,10 @@ export class CollectionService implements OnModuleInit {
     ) {}
 
     onModuleInit() {
-        this.eventBus
-            .ofType(CatalogModificationEvent)
+        const productEvents$ = this.eventBus.ofType(ProductEvent);
+        const variantEvents$ = this.eventBus.ofType(ProductVariantEvent);
+
+        merge(productEvents$, variantEvents$)
             .pipe(debounceTime(50))
             .subscribe(async event => {
                 const collections = await this.connection.getRepository(Collection).find({

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

@@ -23,7 +23,7 @@ import { ProductVariantTranslation } from '../../entity/product-variant/product-
 import { ProductVariant } from '../../entity/product-variant/product-variant.entity';
 import { Product } from '../../entity/product/product.entity';
 import { EventBus } from '../../event-bus/event-bus';
-import { CatalogModificationEvent } from '../../event-bus/events/catalog-modification-event';
+import { ProductVariantEvent } from '../../event-bus/events/product-variant-event';
 import { ListQueryBuilder } from '../helpers/list-query-builder/list-query-builder';
 import { TaxCalculator } from '../helpers/tax-calculator/tax-calculator';
 import { TranslatableSaver } from '../helpers/translatable-saver/translatable-saver';
@@ -69,6 +69,30 @@ export class ProductVariantService {
             });
     }
 
+    findByIds(ctx: RequestContext, ids: ID[]): Promise<Array<Translated<ProductVariant>>> {
+        return this.connection.manager
+            .getRepository(ProductVariant)
+            .findByIds(ids, {
+                relations: [
+                    'options',
+                    'facetValues',
+                    'facetValues.facet',
+                    'taxCategory',
+                    'assets',
+                    'featuredAsset',
+                ],
+            })
+            .then(variants => {
+                return variants.map(variant =>
+                    translateDeep(this.applyChannelPriceAndTax(variant, ctx), ctx.languageCode, [
+                        'options',
+                        'facetValues',
+                        ['facetValues', 'facet'],
+                    ]),
+                );
+            });
+    }
+
     getVariantsByProductId(ctx: RequestContext, productId: ID): Promise<Array<Translated<ProductVariant>>> {
         return this.connection
             .getRepository(ProductVariant)
@@ -145,7 +169,33 @@ export class ProductVariantService {
             );
     }
 
-    async create(ctx: RequestContext, input: CreateProductVariantInput): Promise<Translated<ProductVariant>> {
+    async create(
+        ctx: RequestContext,
+        input: CreateProductVariantInput[],
+    ): Promise<Array<Translated<ProductVariant>>> {
+        const ids: ID[] = [];
+        for (const productInput of input) {
+            const id = await this.createSingle(ctx, productInput);
+            ids.push(id);
+        }
+        const createdVariants = await this.findByIds(ctx, ids);
+        this.eventBus.publish(new ProductVariantEvent(ctx, createdVariants, 'updated'));
+        return createdVariants;
+    }
+
+    async update(
+        ctx: RequestContext,
+        input: UpdateProductVariantInput[],
+    ): Promise<Array<Translated<ProductVariant>>> {
+        for (const productInput of input) {
+            await this.updateSingle(ctx, productInput);
+        }
+        const updatedVariants = await this.findByIds(ctx, input.map(i => i.id));
+        this.eventBus.publish(new ProductVariantEvent(ctx, updatedVariants, 'updated'));
+        return updatedVariants;
+    }
+
+    private async createSingle(ctx: RequestContext, input: CreateProductVariantInput): Promise<ID> {
         await this.validateVariantOptionIds(ctx, input);
         if (!input.optionIds) {
             input.optionIds = [];
@@ -192,11 +242,10 @@ export class ProductVariantService {
         }
 
         await this.createProductVariantPrice(createdVariant.id, createdVariant.price, ctx.channelId);
-        this.eventBus.publish(new CatalogModificationEvent(ctx, createdVariant));
-        return await assertFound(this.findOne(ctx, createdVariant.id));
+        return createdVariant.id;
     }
 
-    async update(ctx: RequestContext, input: UpdateProductVariantInput): Promise<Translated<ProductVariant>> {
+    private async updateSingle(ctx: RequestContext, input: UpdateProductVariantInput): Promise<ID> {
         const existingVariant = await getEntityOrThrow(this.connection, ProductVariant, input.id);
         if (input.stockOnHand && input.stockOnHand < 0) {
             throw new UserInputError('error.stockonhand-cannot-be-negative');
@@ -244,24 +293,7 @@ export class ProductVariantService {
             variantPrice.price = input.price;
             await variantPriceRepository.save(variantPrice);
         }
-        const variant = await assertFound(
-            this.connection.manager.getRepository(ProductVariant).findOne(input.id, {
-                relations: [
-                    'options',
-                    'facetValues',
-                    'facetValues.facet',
-                    'taxCategory',
-                    'assets',
-                    'featuredAsset',
-                ],
-            }),
-        );
-        this.eventBus.publish(new CatalogModificationEvent(ctx, variant));
-        return translateDeep(this.applyChannelPriceAndTax(variant, ctx), ctx.languageCode, [
-            'options',
-            'facetValues',
-            ['facetValues', 'facet'],
-        ]);
+        return existingVariant.id;
     }
 
     /**
@@ -284,7 +316,7 @@ export class ProductVariantService {
         const variant = await getEntityOrThrow(this.connection, ProductVariant, id);
         variant.deletedAt = new Date();
         await this.connection.getRepository(ProductVariant).save(variant);
-        this.eventBus.publish(new CatalogModificationEvent(ctx, variant));
+        this.eventBus.publish(new ProductVariantEvent(ctx, [variant], 'deleted'));
         return {
             result: DeletionResult.DELETED,
         };

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

@@ -23,7 +23,8 @@ import { ProductOptionGroup } from '../../entity/product-option-group/product-op
 import { ProductTranslation } from '../../entity/product/product-translation.entity';
 import { Product } from '../../entity/product/product.entity';
 import { EventBus } from '../../event-bus/event-bus';
-import { CatalogModificationEvent } from '../../event-bus/events/catalog-modification-event';
+import { ProductChannelEvent } from '../../event-bus/events/product-channel-event';
+import { ProductEvent } from '../../event-bus/events/product-event';
 import { ListQueryBuilder } from '../helpers/list-query-builder/list-query-builder';
 import { TranslatableSaver } from '../helpers/translatable-saver/translatable-saver';
 import { findByIdsInChannel, findOneInChannel } from '../helpers/utils/channel-aware-orm-utils';
@@ -136,7 +137,7 @@ export class ProductService {
             },
         });
         await this.assetService.updateEntityAssets(product, input);
-        this.eventBus.publish(new CatalogModificationEvent(ctx, product));
+        this.eventBus.publish(new ProductEvent(ctx, product, 'created'));
         return assertFound(this.findOne(ctx, product.id));
     }
 
@@ -155,7 +156,7 @@ export class ProductService {
                 await this.assetService.updateEntityAssets(p, input);
             },
         });
-        this.eventBus.publish(new CatalogModificationEvent(ctx, product));
+        this.eventBus.publish(new ProductEvent(ctx, product, 'updated'));
         return assertFound(this.findOne(ctx, product.id));
     }
 
@@ -163,7 +164,7 @@ export class ProductService {
         const product = await getEntityOrThrow(this.connection, Product, productId, ctx.channelId);
         product.deletedAt = new Date();
         await this.connection.getRepository(Product).save(product);
-        this.eventBus.publish(new CatalogModificationEvent(ctx, product));
+        this.eventBus.publish(new ProductEvent(ctx, product, 'deleted'));
         return {
             result: DeletionResult.DELETED,
         };
@@ -196,7 +197,7 @@ export class ProductService {
                     input.channelId,
                 );
             }
-            this.eventBus.publish(new CatalogModificationEvent(ctx, product));
+            this.eventBus.publish(new ProductChannelEvent(ctx, product, input.channelId, 'assigned'));
         }
         return this.findByIds(ctx, productsWithVariants.map(p => p.id));
     }
@@ -219,7 +220,7 @@ export class ProductService {
         const products = await this.connection.getRepository(Product).findByIds(input.productIds);
         for (const product of products) {
             await this.channelService.removeFromChannels(Product, product.id, [input.channelId]);
-            this.eventBus.publish(new CatalogModificationEvent(ctx, product));
+            this.eventBus.publish(new ProductChannelEvent(ctx, product, input.channelId, 'removed'));
         }
         return this.findByIds(ctx, products.map(p => p.id));
     }

packages/core/src/worker/types.ts

@@ -34,5 +34,5 @@
 export abstract class WorkerMessage<T, R> {
     static readonly pattern: string;
     constructor(public data: T) {}
-    response?: R;
+    response: R;
 }