refactor(core): Get DefaultSearchPlugin working with variant channels

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

@@ -15,6 +15,7 @@ import { testConfig, TEST_SETUP_TIMEOUT_MS } from '../../../e2e-common/test-conf
 
 import {
     AssignProductsToChannel,
+    AssignProductVariantsToChannel,
     ChannelFragment,
     CreateChannel,
     CreateCollection,
@@ -26,6 +27,7 @@ import {
     LanguageCode,
     Reindex,
     RemoveProductsFromChannel,
+    RemoveProductVariantsFromChannel,
     SearchFacetValues,
     SearchGetAssets,
     SearchGetPrices,
@@ -40,6 +42,7 @@ import {
 } from './graphql/generated-e2e-admin-types';
 import { LogicalOperator, SearchProductsShop } from './graphql/generated-e2e-shop-types';
 import {
+    ASSIGN_PRODUCTVARIANT_TO_CHANNEL,
     ASSIGN_PRODUCT_TO_CHANNEL,
     CREATE_CHANNEL,
     CREATE_COLLECTION,
@@ -47,6 +50,7 @@ import {
     DELETE_ASSET,
     DELETE_PRODUCT,
     DELETE_PRODUCT_VARIANT,
+    REMOVE_PRODUCTVARIANT_FROM_CHANNEL,
     REMOVE_PRODUCT_FROM_CHANNEL,
     UPDATE_ASSET,
     UPDATE_COLLECTION,
@@ -886,7 +890,7 @@ describe('Default search plugin', () => {
                         defaultShippingZoneId: 'T_1',
                     },
                 });
-                secondChannel = createChannel;
+                secondChannel = createChannel as ChannelFragment;
             });
 
             it('adding product to channel', async () => {
@@ -921,6 +925,56 @@ describe('Default search plugin', () => {
                 const { search } = await doAdminSearchQuery({ groupByProduct: true });
                 expect(search.items.map(i => i.productId)).toEqual(['T_1']);
             }, 10000);
+
+            it('adding product variant to channel', async () => {
+                adminClient.setChannelToken(E2E_DEFAULT_CHANNEL_TOKEN);
+                await adminClient.query<
+                    AssignProductVariantsToChannel.Mutation,
+                    AssignProductVariantsToChannel.Variables
+                >(ASSIGN_PRODUCTVARIANT_TO_CHANNEL, {
+                    input: { channelId: secondChannel.id, productVariantIds: ['T_10', 'T_15'] },
+                });
+                await awaitRunningJobs(adminClient);
+
+                adminClient.setChannelToken(SECOND_CHANNEL_TOKEN);
+
+                const { search: searchGrouped } = await doAdminSearchQuery({ groupByProduct: true });
+                expect(searchGrouped.items.map(i => i.productId)).toEqual(['T_1', 'T_3', 'T_4']);
+
+                const { search: searchUngrouped } = await doAdminSearchQuery({ groupByProduct: false });
+                expect(searchUngrouped.items.map(i => i.productVariantId)).toEqual([
+                    'T_1',
+                    'T_2',
+                    'T_3',
+                    'T_4',
+                    'T_10',
+                    'T_15',
+                ]);
+            }, 10000);
+
+            it('removing product variant to channel', async () => {
+                adminClient.setChannelToken(E2E_DEFAULT_CHANNEL_TOKEN);
+                await adminClient.query<
+                    RemoveProductVariantsFromChannel.Mutation,
+                    RemoveProductVariantsFromChannel.Variables
+                >(REMOVE_PRODUCTVARIANT_FROM_CHANNEL, {
+                    input: { channelId: secondChannel.id, productVariantIds: ['T_1', 'T_15'] },
+                });
+                await awaitRunningJobs(adminClient);
+
+                adminClient.setChannelToken(SECOND_CHANNEL_TOKEN);
+
+                const { search: searchGrouped } = await doAdminSearchQuery({ groupByProduct: true });
+                expect(searchGrouped.items.map(i => i.productId)).toEqual(['T_1', 'T_3']);
+
+                const { search: searchUngrouped } = await doAdminSearchQuery({ groupByProduct: false });
+                expect(searchUngrouped.items.map(i => i.productVariantId)).toEqual([
+                    'T_2',
+                    'T_3',
+                    'T_4',
+                    'T_10',
+                ]);
+            }, 10000);
         });
 
         describe('multiple language handling', () => {

packages/core/src/plugin/default-search-plugin/default-search-plugin.ts

@@ -8,6 +8,7 @@ import { AssetEvent } from '../../event-bus/events/asset-event';
 import { CollectionModificationEvent } from '../../event-bus/events/collection-modification-event';
 import { ProductChannelEvent } from '../../event-bus/events/product-channel-event';
 import { ProductEvent } from '../../event-bus/events/product-event';
+import { ProductVariantChannelEvent } from '../../event-bus/events/product-variant-channel-event';
 import { ProductVariantEvent } from '../../event-bus/events/product-variant-event';
 import { TaxRateModificationEvent } from '../../event-bus/events/tax-rate-modification-event';
 import { PluginCommonModule } from '../plugin-common.module';
@@ -106,6 +107,21 @@ export class DefaultSearchPlugin implements OnVendureBootstrap {
                 );
             }
         });
+        this.eventBus.ofType(ProductVariantChannelEvent).subscribe(event => {
+            if (event.type === 'assigned') {
+                return this.searchIndexService.assignVariantToChannel(
+                    event.ctx,
+                    event.productVariant.id,
+                    event.channelId,
+                );
+            } else {
+                return this.searchIndexService.removeVariantFromChannel(
+                    event.ctx,
+                    event.productVariant.id,
+                    event.channelId,
+                );
+            }
+        });
 
         const collectionModification$ = this.eventBus.ofType(CollectionModificationEvent);
         const closingNotifier$ = collectionModification$.pipe(debounceTime(50));

packages/core/src/plugin/default-search-plugin/indexer/indexer.controller.ts

@@ -21,11 +21,13 @@ import { asyncObservable } from '../../../worker/async-observable';
 import { SearchIndexItem } from '../search-index-item.entity';
 import {
     AssignProductToChannelMessage,
+    AssignVariantToChannelMessage,
     DeleteAssetMessage,
     DeleteProductMessage,
     DeleteVariantMessage,
     ReindexMessage,
     RemoveProductFromChannelMessage,
+    RemoveVariantFromChannelMessage,
     UpdateAssetMessage,
     UpdateProductMessage,
     UpdateVariantMessage,
@@ -44,6 +46,7 @@ export const variantRelations = [
     'facetValues.facet',
     'collections',
     'taxCategory',
+    'channels',
 ];
 
 export const workerLoggerCtx = 'DefaultSearchPlugin Worker';
@@ -198,6 +201,27 @@ export class IndexerController {
         });
     }
 
+    @MessagePattern(AssignVariantToChannelMessage.pattern)
+    assignVariantToChannel(
+        data: AssignVariantToChannelMessage['data'],
+    ): Observable<AssignProductToChannelMessage['response']> {
+        const ctx = RequestContext.deserialize(data.ctx);
+        return asyncObservable(async () => {
+            return this.updateVariantsInChannel(ctx, [data.productVariantId], data.channelId);
+        });
+    }
+
+    @MessagePattern(RemoveVariantFromChannelMessage.pattern)
+    removeVariantFromChannel(
+        data: RemoveVariantFromChannelMessage['data'],
+    ): Observable<RemoveProductFromChannelMessage['response']> {
+        const ctx = RequestContext.deserialize(data.ctx);
+        return asyncObservable(async () => {
+            await this.removeSearchIndexItems(ctx.languageCode, data.channelId, [data.productVariantId]);
+            return true;
+        });
+    }
+
     @MessagePattern(UpdateAssetMessage.pattern)
     updateAsset(data: UpdateAssetMessage['data']): Observable<UpdateAssetMessage['response']> {
         return asyncObservable(async () => {

packages/core/src/plugin/default-search-plugin/indexer/search-index.service.ts

@@ -14,12 +14,14 @@ import { WorkerMessage } from '../../../worker/types';
 import { WorkerService } from '../../../worker/worker.service';
 import {
     AssignProductToChannelMessage,
+    AssignVariantToChannelMessage,
     DeleteAssetMessage,
     DeleteProductMessage,
     DeleteVariantMessage,
     ReindexMessage,
     ReindexMessageResponse,
     RemoveProductFromChannelMessage,
+    RemoveVariantFromChannelMessage,
     UpdateAssetMessage,
     UpdateIndexQueueJobData,
     UpdateProductMessage,
@@ -40,7 +42,7 @@ export class SearchIndexService {
         updateIndexQueue = this.jobService.createQueue({
             name: 'update-search-index',
             concurrency: 1,
-            process: (job) => {
+            process: job => {
                 const data = job.data;
                 switch (data.type) {
                     case 'reindex':
@@ -74,6 +76,12 @@ export class SearchIndexService {
                     case 'remove-product-from-channel':
                         this.sendMessage(job, new RemoveProductFromChannelMessage(data));
                         break;
+                    case 'assign-variant-to-channel':
+                        this.sendMessage(job, new AssignVariantToChannelMessage(data));
+                        break;
+                    case 'remove-variant-from-channel':
+                        this.sendMessage(job, new RemoveVariantFromChannelMessage(data));
+                        break;
                     default:
                         assertNever(data);
                 }
@@ -90,7 +98,7 @@ export class SearchIndexService {
     }
 
     updateVariants(ctx: RequestContext, variants: ProductVariant[]) {
-        const variantIds = variants.map((v) => v.id);
+        const variantIds = variants.map(v => v.id);
         this.addJobToQueue({ type: 'update-variants', ctx: ctx.serialize(), variantIds });
     }
 
@@ -99,7 +107,7 @@ export class SearchIndexService {
     }
 
     deleteVariant(ctx: RequestContext, variants: ProductVariant[]) {
-        const variantIds = variants.map((v) => v.id);
+        const variantIds = variants.map(v => v.id);
         this.addJobToQueue({ type: 'delete-variant', ctx: ctx.serialize(), variantIds });
     }
 
@@ -133,6 +141,24 @@ export class SearchIndexService {
         });
     }
 
+    assignVariantToChannel(ctx: RequestContext, productVariantId: ID, channelId: ID) {
+        this.addJobToQueue({
+            type: 'assign-variant-to-channel',
+            ctx: ctx.serialize(),
+            productVariantId,
+            channelId,
+        });
+    }
+
+    removeVariantFromChannel(ctx: RequestContext, productVariantId: ID, channelId: ID) {
+        this.addJobToQueue({
+            type: 'remove-variant-from-channel',
+            ctx: ctx.serialize(),
+            productVariantId,
+            channelId,
+        });
+    }
+
     private addJobToQueue(data: UpdateIndexQueueJobData) {
         if (updateIndexQueue) {
             return updateIndexQueue.add(data);
@@ -142,7 +168,7 @@ export class SearchIndexService {
     private sendMessage(job: Job<any>, message: WorkerMessage<any, any>) {
         this.workerService.send(message).subscribe({
             complete: () => job.complete(true),
-            error: (err) => {
+            error: err => {
                 Logger.error(err);
                 job.fail(err);
             },

packages/core/src/plugin/default-search-plugin/types.ts

@@ -40,6 +40,12 @@ export type ProductChannelMessageData = {
     channelId: ID;
 };
 
+export type VariantChannelMessageData = {
+    ctx: SerializedRequestContext;
+    productVariantId: ID;
+    channelId: ID;
+};
+
 export class ReindexMessage extends WorkerMessage<ReindexMessageData, ReindexMessageResponse> {
     static readonly pattern = 'Reindex';
 }
@@ -67,6 +73,12 @@ export class AssignProductToChannelMessage extends WorkerMessage<ProductChannelM
 export class RemoveProductFromChannelMessage extends WorkerMessage<ProductChannelMessageData, boolean> {
     static readonly pattern = 'RemoveProductFromChannel';
 }
+export class AssignVariantToChannelMessage extends WorkerMessage<VariantChannelMessageData, boolean> {
+    static readonly pattern = 'AssignVariantToChannel';
+}
+export class RemoveVariantFromChannelMessage extends WorkerMessage<VariantChannelMessageData, boolean> {
+    static readonly pattern = 'RemoveVariantFromChannel';
+}
 export class UpdateAssetMessage extends WorkerMessage<UpdateAssetMessageData, boolean> {
     static readonly pattern = 'UpdateAsset';
 }
@@ -86,6 +98,8 @@ 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>;
+type AssignVariantToChannelJobData = NamedJobData<'assign-variant-to-channel', VariantChannelMessageData>;
+type RemoveVariantFromChannelJobData = NamedJobData<'remove-variant-from-channel', VariantChannelMessageData>;
 export type UpdateIndexQueueJobData =
     | ReindexJobData
     | UpdateProductJobData
@@ -96,4 +110,6 @@ export type UpdateIndexQueueJobData =
     | UpdateAssetJobData
     | DeleteAssetJobData
     | AssignProductToChannelJobData
-    | RemoveProductFromChannelJobData;
+    | RemoveProductFromChannelJobData
+    | AssignVariantToChannelJobData
+    | RemoveVariantFromChannelJobData;

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

@@ -24,6 +24,7 @@ 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 { 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 { SlugValidator } from '../helpers/slug-validator/slug-validator';
@@ -211,6 +212,10 @@ export class ProductService {
             channelId: input.channelId,
             priceFactor: input.priceFactor,
         });
+        const products = await this.connection.getRepository(ctx, Product).findByIds(input.productIds);
+        for (const product of products) {
+            this.eventBus.publish(new ProductChannelEvent(ctx, product, input.channelId, 'assigned'));
+        }
         return this.findByIds(
             ctx,
             productsWithVariants.map(p => p.id),
@@ -232,6 +237,10 @@ export class ProductService {
             ),
             channelId: input.channelId,
         });
+        const products = await this.connection.getRepository(ctx, Product).findByIds(input.productIds);
+        for (const product of products) {
+            this.eventBus.publish(new ProductChannelEvent(ctx, product, input.channelId, 'removed'));
+        }
         return this.findByIds(
             ctx,
             productsWithVariants.map(p => p.id),