refactor: Await EventBus.publish() method calls

Relates to #2735

packages/core/e2e/fixtures/test-plugins/transaction-test-plugin.ts

@@ -20,7 +20,10 @@ import { ReplaySubject, Subscription } from 'rxjs';
 import { vi } from 'vitest';
 
 export class TestEvent extends VendureEvent {
-    constructor(public ctx: RequestContext, public administrator: Administrator) {
+    constructor(
+        public ctx: RequestContext,
+        public administrator: Administrator,
+    ) {
         super();
     }
 }
@@ -58,7 +61,10 @@ class TestUserService {
 
 @Injectable()
 class TestAdminService {
-    constructor(private connection: TransactionalConnection, private userService: TestUserService) {}
+    constructor(
+        private connection: TransactionalConnection,
+        private userService: TestUserService,
+    ) {}
 
     async createAdministrator(ctx: RequestContext, emailAddress: string, fail: boolean) {
         const user = await this.userService.createUser(ctx, emailAddress);
@@ -91,7 +97,7 @@ class TestResolver {
     @Transaction()
     async createTestAdministrator(@Ctx() ctx: RequestContext, @Args() args: any) {
         const admin = await this.testAdminService.createAdministrator(ctx, args.emailAddress, args.fail);
-        this.eventBus.publish(new TestEvent(ctx, admin));
+        await this.eventBus.publish(new TestEvent(ctx, admin));
         return admin;
     }
 
@@ -113,7 +119,7 @@ class TestResolver {
     @Transaction()
     async createTestAdministrator4(@Ctx() ctx: RequestContext, @Args() args: any) {
         const admin = await this.testAdminService.createAdministrator(ctx, args.emailAddress, args.fail);
-        this.eventBus.publish(new TestEvent(ctx, admin));
+        await this.eventBus.publish(new TestEvent(ctx, admin));
         await new Promise(resolve => setTimeout(resolve, 50));
         return admin;
     }
@@ -157,8 +163,8 @@ class TestResolver {
                             i < args.n * args.failFactor,
                         ),
                     )
-                    .then(admin => {
-                        this.eventBus.publish(new TestEvent(ctx, admin));
+                    .then(async admin => {
+                        await this.eventBus.publish(new TestEvent(ctx, admin));
                         return admin;
                     }),
             );
@@ -310,7 +316,10 @@ export class TransactionTestPlugin implements OnApplicationBootstrap {
     static errorHandler = vi.fn();
     static eventHandlerComplete$ = new ReplaySubject(1);
 
-    constructor(private eventBus: EventBus, private connection: TransactionalConnection) {}
+    constructor(
+        private eventBus: EventBus,
+        private connection: TransactionalConnection,
+    ) {}
 
     static reset() {
         this.eventHandlerComplete$ = new ReplaySubject(1);

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

@@ -105,6 +105,11 @@ export class EventBus implements OnModuleDestroy {
     /**
      * @description
      * Publish an event which any subscribers can react to.
+     *
+     * @example
+     * ```ts
+     * await eventBus.publish(new SomeEvent());
+     * ```
      */
     async publish<T extends VendureEvent>(event: T): Promise<void> {
         this.eventStream.next(event);

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

@@ -55,7 +55,7 @@ export class FulltextSearchService {
     ): Promise<Omit<Omit<SearchResponse, 'facetValues'>, 'collections'>> {
         const items = await this._searchStrategy.getSearchResults(ctx, input, enabledOnly);
         const totalItems = await this._searchStrategy.getTotalCount(ctx, input, enabledOnly);
-        this.eventBus.publish(new SearchEvent(ctx, input));
+        await this.eventBus.publish(new SearchEvent(ctx, input));
 
         return {
             items,

packages/core/src/service/helpers/order-modifier/order-modifier.ts

@@ -207,7 +207,7 @@ export class OrderModifier {
         );
         order.lines.push(lineWithRelations);
         await this.connection.getRepository(ctx, Order).save(order, { reload: false });
-        this.eventBus.publish(new OrderLineEvent(ctx, order, lineWithRelations, 'created'));
+        await this.eventBus.publish(new OrderLineEvent(ctx, order, lineWithRelations, 'created'));
         return lineWithRelations;
     }
 
@@ -250,7 +250,7 @@ export class OrderModifier {
             }
         }
         await this.connection.getRepository(ctx, OrderLine).save(orderLine);
-        this.eventBus.publish(new OrderLineEvent(ctx, order, orderLine, 'updated'));
+        await this.eventBus.publish(new OrderLineEvent(ctx, order, orderLine, 'updated'));
         return orderLine;
     }
 
@@ -403,8 +403,8 @@ export class OrderModifier {
         const refundInputArray = Array.isArray(input.refunds)
             ? input.refunds
             : input.refund
-            ? [input.refund]
-            : [];
+              ? [input.refund]
+              : [];
         const refundInputs: RefundOrderInput[] = refundInputArray.map(refund => ({
             lines: [],
             adjustment: 0,

packages/core/src/service/initializer.service.ts

@@ -53,7 +53,7 @@ export class InitializerService {
         await this.shippingMethodService.initShippingMethods();
         await this.taxRateService.initTaxRates();
         await this.stockLocationService.initStockLocations();
-        this.eventBus.publish(new InitializerEvent());
+        await this.eventBus.publish(new InitializerEvent());
     }
 
     /**

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

@@ -141,7 +141,7 @@ export class AdministratorService {
             input,
             createdAdministrator,
         );
-        this.eventBus.publish(new AdministratorEvent(ctx, createdAdministrator, 'created', input));
+        await this.eventBus.publish(new AdministratorEvent(ctx, createdAdministrator, 'created', input));
         return createdAdministrator;
     }
 
@@ -193,8 +193,8 @@ export class AdministratorService {
             for (const roleId of input.roleIds) {
                 updatedAdministrator = await this.assignRole(ctx, administrator.id, roleId);
             }
-            this.eventBus.publish(new RoleChangeEvent(ctx, administrator, addIds, 'assigned'));
-            this.eventBus.publish(new RoleChangeEvent(ctx, administrator, removeIds, 'removed'));
+            await this.eventBus.publish(new RoleChangeEvent(ctx, administrator, addIds, 'assigned'));
+            await this.eventBus.publish(new RoleChangeEvent(ctx, administrator, removeIds, 'removed'));
         }
         await this.customFieldRelationService.updateRelations(
             ctx,
@@ -202,7 +202,7 @@ export class AdministratorService {
             input,
             updatedAdministrator,
         );
-        this.eventBus.publish(new AdministratorEvent(ctx, administrator, 'updated', input));
+        await this.eventBus.publish(new AdministratorEvent(ctx, administrator, 'updated', input));
         return updatedAdministrator;
     }
 
@@ -262,7 +262,7 @@ export class AdministratorService {
         await this.connection.getRepository(ctx, Administrator).update({ id }, { deletedAt: new Date() });
         // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
         await this.userService.softDelete(ctx, administrator.user.id);
-        this.eventBus.publish(new AdministratorEvent(ctx, administrator, 'deleted', id));
+        await this.eventBus.publish(new AdministratorEvent(ctx, administrator, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
         };

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

@@ -311,7 +311,7 @@ export class AssetService {
                 result.tags = tags;
                 await this.connection.getRepository(ctx, Asset).save(result);
             }
-            this.eventBus.publish(new AssetEvent(ctx, result, 'created', input));
+            await this.eventBus.publish(new AssetEvent(ctx, result, 'created', input));
             resolve(result);
         });
     }
@@ -333,7 +333,7 @@ export class AssetService {
             asset.tags = await this.tagService.valuesToTags(ctx, input.tags);
         }
         const updatedAsset = await this.connection.getRepository(ctx, Asset).save(asset);
-        this.eventBus.publish(new AssetEvent(ctx, updatedAsset, 'updated', input));
+        await this.eventBus.publish(new AssetEvent(ctx, updatedAsset, 'updated', input));
         return updatedAsset;
     }
 
@@ -385,7 +385,7 @@ export class AssetService {
             await Promise.all(
                 assets.map(async asset => {
                     await this.channelService.removeFromChannels(ctx, Asset, asset.id, [ctx.channelId]);
-                    this.eventBus.publish(new AssetChannelEvent(ctx, asset, ctx.channelId, 'removed'));
+                    await this.eventBus.publish(new AssetChannelEvent(ctx, asset, ctx.channelId, 'removed'));
                 }),
             );
             const isOnlyChannel = channelsOfAssets.length === 1;
@@ -401,7 +401,7 @@ export class AssetService {
         await Promise.all(
             assets.map(async asset => {
                 await this.channelService.removeFromChannels(ctx, Asset, asset.id, channelsOfAssets);
-                this.eventBus.publish(new AssetChannelEvent(ctx, asset, ctx.channelId, 'removed'));
+                await this.eventBus.publish(new AssetChannelEvent(ctx, asset, ctx.channelId, 'removed'));
             }),
         );
         return this.deleteUnconditional(ctx, assets);
@@ -426,7 +426,9 @@ export class AssetService {
         await Promise.all(
             assets.map(async asset => {
                 await this.channelService.assignToChannels(ctx, Asset, asset.id, [input.channelId]);
-                return this.eventBus.publish(new AssetChannelEvent(ctx, asset, input.channelId, 'assigned'));
+                return await this.eventBus.publish(
+                    new AssetChannelEvent(ctx, asset, input.channelId, 'assigned'),
+                );
             }),
         );
         return this.connection.findByIdsInChannel(
@@ -465,8 +467,8 @@ export class AssetService {
                 maybeFilePathOrCtx instanceof RequestContext
                     ? maybeFilePathOrCtx
                     : maybeCtx instanceof RequestContext
-                    ? maybeCtx
-                    : RequestContext.empty();
+                      ? maybeCtx
+                      : RequestContext.empty();
             return this.createAssetInternal(ctx, stream, filename, mimetype);
         } else {
             throw new InternalServerError('error.path-should-be-a-string-got-buffer');
@@ -500,7 +502,7 @@ export class AssetService {
             } catch (e: any) {
                 Logger.error('error.could-not-delete-asset-file', undefined, e.stack);
             }
-            this.eventBus.publish(new AssetEvent(ctx, deletedAsset, 'deleted', deletedAsset.id));
+            await this.eventBus.publish(new AssetEvent(ctx, deletedAsset, 'deleted', deletedAsset.id));
         }
         return {
             result: DeletionResult.DELETED,

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

@@ -52,7 +52,7 @@ export class AuthService {
         authenticationMethod: string,
         authenticationData: any,
     ): Promise<AuthenticatedSession | InvalidCredentialsError | NotVerifiedError> {
-        this.eventBus.publish(
+        await this.eventBus.publish(
             new AttemptedLoginEvent(
                 ctx,
                 authenticationMethod,
@@ -104,7 +104,7 @@ export class AuthService {
             user,
             authenticationStrategyName,
         );
-        this.eventBus.publish(new LoginEvent(ctx, user));
+        await this.eventBus.publish(new LoginEvent(ctx, user));
         return session;
     }
 
@@ -147,7 +147,7 @@ export class AuthService {
             if (typeof authenticationStrategy.onLogOut === 'function') {
                 await authenticationStrategy.onLogOut(ctx, session.user);
             }
-            this.eventBus.publish(new LogoutEvent(ctx));
+            await this.eventBus.publish(new LogoutEvent(ctx));
             return this.sessionService.deleteSessionsByUser(ctx, session.user);
         }
     }

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

@@ -16,7 +16,12 @@ import { FindOptionsWhere } from 'typeorm';
 import { RelationPaths } from '../../api';
 import { RequestContext } from '../../api/common/request-context';
 import { ErrorResultUnion, isGraphQlErrorResult } from '../../common/error/error-result';
-import { ChannelNotFoundError, EntityNotFoundError, InternalServerError, UserInputError } from '../../common/error/errors';
+import {
+    ChannelNotFoundError,
+    EntityNotFoundError,
+    InternalServerError,
+    UserInputError,
+} from '../../common/error/errors';
 import { LanguageNotAvailableError } from '../../common/error/generated-graphql-admin-errors';
 import { createSelfRefreshingCache, SelfRefreshingCache } from '../../common/self-refreshing-cache';
 import { ChannelAware, ListQueryOptions } from '../../common/types/common-types';
@@ -115,7 +120,7 @@ export class ChannelService {
         const defaultChannel = await this.getDefaultChannel(ctx);
         const channelIds = unique([ctx.channelId, defaultChannel.id]);
         entity.channels = channelIds.map(id => ({ id })) as any;
-        this.eventBus.publish(new ChangeChannelEvent(ctx, entity, [ctx.channelId], 'assigned'));
+        await this.eventBus.publish(new ChangeChannelEvent(ctx, entity, [ctx.channelId], 'assigned'));
         return entity;
     }
 
@@ -130,7 +135,11 @@ export class ChannelService {
      * @returns A promise that resolves to an array of objects, each containing a channel ID.
      * @private
      */
-    private async getAssignedEntityChannels<T extends ChannelAware & VendureEntity>(ctx: RequestContext, entityType: Type<T>, entityId: T['id']): Promise<{ channelId: ID }[]> {
+    private async getAssignedEntityChannels<T extends ChannelAware & VendureEntity>(
+        ctx: RequestContext,
+        entityType: Type<T>,
+        entityId: T['id'],
+    ): Promise<Array<{ channelId: ID }>> {
         const repository = this.connection.getRepository(ctx, entityType);
 
         const metadata = repository.metadata;
@@ -142,13 +151,18 @@ export class ChannelService {
 
         const junctionTableName = channelsRelation.junctionEntityMetadata?.tableName;
         const junctionColumnName = channelsRelation.junctionEntityMetadata?.columns[0].databaseName;
-        const inverseJunctionColumnName = channelsRelation.junctionEntityMetadata?.inverseColumns[0].databaseName;
+        const inverseJunctionColumnName =
+            channelsRelation.junctionEntityMetadata?.inverseColumns[0].databaseName;
 
         if (!junctionTableName || !junctionColumnName || !inverseJunctionColumnName) {
-            throw new InternalServerError(`Could not find necessary join table information for the channels relation of entity ${metadata.name}`);
+            throw new InternalServerError(
+                `Could not find necessary join table information for the channels relation of entity ${metadata.name}`,
+            );
         }
 
-        return await this.connection.getRepository(ctx, entityType).createQueryBuilder()
+        return await this.connection
+            .getRepository(ctx, entityType)
+            .createQueryBuilder()
             .select(`channel.${inverseJunctionColumnName}`, 'channelId')
             .from(junctionTableName, 'channel')
             .where(`channel.${junctionColumnName} = :entityId`, { entityId })
@@ -183,7 +197,9 @@ export class ChannelService {
         });
         const assignedChannels = await this.getAssignedEntityChannels(ctx, entityType, entityId);
 
-        const newChannelIds = channelIds.filter(id => !assignedChannels.some(ec => idsAreEqual(ec.channelId, id)))
+        const newChannelIds = channelIds.filter(
+            id => !assignedChannels.some(ec => idsAreEqual(ec.channelId, id)),
+        );
 
         await this.connection
             .getRepository(ctx, entityType)
@@ -192,7 +208,7 @@ export class ChannelService {
             .of(entity.id)
             .add(newChannelIds);
 
-        this.eventBus.publish(new ChangeChannelEvent(ctx, entity, channelIds, 'assigned', entityType));
+        await this.eventBus.publish(new ChangeChannelEvent(ctx, entity, channelIds, 'assigned', entityType));
         return entity;
     }
 
@@ -212,16 +228,18 @@ export class ChannelService {
             where: {
                 id: entityId,
             } as FindOptionsWhere<T>,
-        })
+        });
         if (!entity) {
             return;
         }
         const assignedChannels = await this.getAssignedEntityChannels(ctx, entityType, entityId);
 
-        const existingChannelIds = channelIds.filter(id => assignedChannels.some(ec => idsAreEqual(ec.channelId, id)));
+        const existingChannelIds = channelIds.filter(id =>
+            assignedChannels.some(ec => idsAreEqual(ec.channelId, id)),
+        );
 
         if (!existingChannelIds.length) {
-            return
+            return;
         }
         await this.connection
             .getRepository(ctx, entityType)
@@ -229,7 +247,7 @@ export class ChannelService {
             .relation('channels')
             .of(entity.id)
             .remove(existingChannelIds);
-        this.eventBus.publish(new ChangeChannelEvent(ctx, entity, channelIds, 'removed', entityType));
+        await this.eventBus.publish(new ChangeChannelEvent(ctx, entity, channelIds, 'removed', entityType));
         return entity;
     }
 
@@ -337,7 +355,7 @@ export class ChannelService {
         }
         await this.customFieldRelationService.updateRelations(ctx, Channel, input, newChannel);
         await this.allChannels.refresh(ctx);
-        this.eventBus.publish(new ChannelEvent(ctx, newChannel, 'created', input));
+        await this.eventBus.publish(new ChannelEvent(ctx, newChannel, 'created', input));
         return newChannel;
     }
 
@@ -433,7 +451,7 @@ export class ChannelService {
         await this.connection.getRepository(ctx, Channel).save(updatedChannel, { reload: false });
         await this.customFieldRelationService.updateRelations(ctx, Channel, input, updatedChannel);
         await this.allChannels.refresh(ctx);
-        this.eventBus.publish(new ChannelEvent(ctx, channel, 'updated', input));
+        await this.eventBus.publish(new ChannelEvent(ctx, channel, 'updated', input));
         return assertFound(this.findOne(ctx, channel.id));
     }
 
@@ -445,7 +463,7 @@ export class ChannelService {
         await this.connection.getRepository(ctx, ProductVariantPrice).delete({
             channelId: id,
         });
-        this.eventBus.publish(new ChannelEvent(ctx, deletedChannel, 'deleted', id));
+        await this.eventBus.publish(new ChannelEvent(ctx, deletedChannel, 'deleted', id));
 
         return {
             result: DeletionResult.DELETED,

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

@@ -146,7 +146,7 @@ export class CollectionService implements OnModuleInit {
                         }
                         job.setProgress(Math.ceil((completed / job.data.collectionIds.length) * 100));
                         if (affectedVariantIds.length) {
-                            this.eventBus.publish(
+                            await this.eventBus.publish(
                                 new CollectionModificationEvent(ctx, collection, affectedVariantIds),
                             );
                         }
@@ -472,7 +472,7 @@ export class CollectionService implements OnModuleInit {
             ctx: ctx.serialize(),
             collectionIds: [collection.id],
         });
-        this.eventBus.publish(new CollectionEvent(ctx, collectionWithRelations, 'created', input));
+        await this.eventBus.publish(new CollectionEvent(ctx, collectionWithRelations, 'created', input));
         return assertFound(this.findOne(ctx, collection.id));
     }
 
@@ -500,9 +500,9 @@ export class CollectionService implements OnModuleInit {
             });
         } else {
             const affectedVariantIds = await this.getCollectionProductVariantIds(collection);
-            this.eventBus.publish(new CollectionModificationEvent(ctx, collection, affectedVariantIds));
+            await this.eventBus.publish(new CollectionModificationEvent(ctx, collection, affectedVariantIds));
         }
-        this.eventBus.publish(new CollectionEvent(ctx, collection, 'updated', input));
+        await this.eventBus.publish(new CollectionEvent(ctx, collection, 'updated', input));
         return assertFound(this.findOne(ctx, collection.id));
     }
 
@@ -526,9 +526,11 @@ export class CollectionService implements OnModuleInit {
                     .remove(chunkedDeleteId);
             }
             await this.connection.getRepository(ctx, Collection).remove(coll);
-            this.eventBus.publish(new CollectionModificationEvent(ctx, deletedColl, affectedVariantIds));
+            await this.eventBus.publish(
+                new CollectionModificationEvent(ctx, deletedColl, affectedVariantIds),
+            );
         }
-        this.eventBus.publish(new CollectionEvent(ctx, deletedCollection, 'deleted', id));
+        await this.eventBus.publish(new CollectionEvent(ctx, deletedCollection, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
         };
@@ -869,7 +871,9 @@ export class CollectionService implements OnModuleInit {
                 await this.channelService.removeFromChannels(ctx, Collection, collection.id, [
                     input.channelId,
                 ]);
-                this.eventBus.publish(new CollectionModificationEvent(ctx, collection, affectedVariantIds));
+                await this.eventBus.publish(
+                    new CollectionModificationEvent(ctx, collection, affectedVariantIds),
+                );
             }),
         );
 

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

@@ -102,7 +102,7 @@ export class CountryService {
             entityType: Country,
             translationType: RegionTranslation,
         });
-        this.eventBus.publish(new CountryEvent(ctx, country, 'created', input));
+        await this.eventBus.publish(new CountryEvent(ctx, country, 'created', input));
         return assertFound(this.findOne(ctx, country.id));
     }
 
@@ -113,7 +113,7 @@ export class CountryService {
             entityType: Country,
             translationType: RegionTranslation,
         });
-        this.eventBus.publish(new CountryEvent(ctx, country, 'updated', input));
+        await this.eventBus.publish(new CountryEvent(ctx, country, 'updated', input));
         return assertFound(this.findOne(ctx, country.id));
     }
 
@@ -133,7 +133,7 @@ export class CountryService {
         } else {
             const deletedCountry = new Country(country);
             await this.connection.getRepository(ctx, Country).remove(country);
-            this.eventBus.publish(new CountryEvent(ctx, deletedCountry, 'deleted', id));
+            await this.eventBus.publish(new CountryEvent(ctx, deletedCountry, 'deleted', id));
             return {
                 result: DeletionResult.DELETED,
                 message: '',

packages/core/src/service/services/customer-group.service.ts

@@ -107,7 +107,7 @@ export class CustomerGroupService {
         }
         const savedCustomerGroup = await assertFound(this.findOne(ctx, newCustomerGroup.id));
         await this.customFieldRelationService.updateRelations(ctx, CustomerGroup, input, savedCustomerGroup);
-        this.eventBus.publish(new CustomerGroupEvent(ctx, savedCustomerGroup, 'created', input));
+        await this.eventBus.publish(new CustomerGroupEvent(ctx, savedCustomerGroup, 'created', input));
         return assertFound(this.findOne(ctx, savedCustomerGroup.id));
     }
 
@@ -121,7 +121,7 @@ export class CustomerGroupService {
             input,
             updatedCustomerGroup,
         );
-        this.eventBus.publish(new CustomerGroupEvent(ctx, customerGroup, 'updated', input));
+        await this.eventBus.publish(new CustomerGroupEvent(ctx, customerGroup, 'updated', input));
         return assertFound(this.findOne(ctx, customerGroup.id));
     }
 
@@ -130,7 +130,7 @@ export class CustomerGroupService {
         try {
             const deletedGroup = new CustomerGroup(group);
             await this.connection.getRepository(ctx, CustomerGroup).remove(group);
-            this.eventBus.publish(new CustomerGroupEvent(ctx, deletedGroup, 'deleted', id));
+            await this.eventBus.publish(new CustomerGroupEvent(ctx, deletedGroup, 'deleted', id));
             return {
                 result: DeletionResult.DELETED,
             };
@@ -163,7 +163,7 @@ export class CustomerGroupService {
         }
 
         await this.connection.getRepository(ctx, Customer).save(customers, { reload: false });
-        this.eventBus.publish(new CustomerGroupChangeEvent(ctx, customers, group, 'assigned'));
+        await this.eventBus.publish(new CustomerGroupChangeEvent(ctx, customers, group, 'assigned'));
 
         return assertFound(this.findOne(ctx, group.id));
     }
@@ -189,7 +189,7 @@ export class CustomerGroupService {
             });
         }
         await this.connection.getRepository(ctx, Customer).save(customers, { reload: false });
-        this.eventBus.publish(new CustomerGroupChangeEvent(ctx, customers, group, 'removed'));
+        await this.eventBus.publish(new CustomerGroupChangeEvent(ctx, customers, group, 'removed'));
         return assertFound(this.findOne(ctx, group.id));
     }
 

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

@@ -268,7 +268,7 @@ export class CustomerService {
                 }
             }
         }
-        this.eventBus.publish(new AccountRegistrationEvent(ctx, customer.user));
+        await this.eventBus.publish(new AccountRegistrationEvent(ctx, customer.user));
         await this.channelService.assignToCurrentChannel(customer, ctx);
         const createdCustomer = await this.connection.getRepository(ctx, Customer).save(customer);
         await this.customFieldRelationService.updateRelations(ctx, Customer, input, createdCustomer);
@@ -291,7 +291,7 @@ export class CustomerService {
                 },
             });
         }
-        this.eventBus.publish(new CustomerEvent(ctx, createdCustomer, 'created', input));
+        await this.eventBus.publish(new CustomerEvent(ctx, createdCustomer, 'created', input));
         return createdCustomer;
     }
 
@@ -364,7 +364,7 @@ export class CustomerService {
                 input,
             },
         });
-        this.eventBus.publish(new CustomerEvent(ctx, customer, 'updated', input));
+        await this.eventBus.publish(new CustomerEvent(ctx, customer, 'updated', input));
         return assertFound(this.findOne(ctx, customer.id));
     }
 
@@ -449,7 +449,7 @@ export class CustomerService {
         await this.connection.getRepository(ctx, User).save(user, { reload: false });
         await this.connection.getRepository(ctx, Customer).save(customer, { reload: false });
         if (!user.verified) {
-            this.eventBus.publish(new AccountRegistrationEvent(ctx, user));
+            await this.eventBus.publish(new AccountRegistrationEvent(ctx, user));
         } else {
             await this.historyService.createHistoryEntryForCustomer({
                 customerId: customer.id,
@@ -472,7 +472,7 @@ export class CustomerService {
         const user = await this.userService.getUserByEmailAddress(ctx, emailAddress);
         if (user && !user.verified) {
             await this.userService.setVerificationToken(ctx, user);
-            this.eventBus.publish(new AccountRegistrationEvent(ctx, user));
+            await this.eventBus.publish(new AccountRegistrationEvent(ctx, user));
         }
     }
 
@@ -506,7 +506,7 @@ export class CustomerService {
             },
         });
         const user = assertFound(this.findOneByUserId(ctx, result.id));
-        this.eventBus.publish(new AccountVerifiedEvent(ctx, customer));
+        await this.eventBus.publish(new AccountVerifiedEvent(ctx, customer));
         return user;
     }
 
@@ -518,7 +518,7 @@ export class CustomerService {
     async requestPasswordReset(ctx: RequestContext, emailAddress: string): Promise<void> {
         const user = await this.userService.setPasswordResetToken(ctx, emailAddress);
         if (user) {
-            this.eventBus.publish(new PasswordResetEvent(ctx, user));
+            await this.eventBus.publish(new PasswordResetEvent(ctx, user));
             const customer = await this.findOneByUserId(ctx, user.id);
             if (!customer) {
                 throw new InternalServerError('error.cannot-locate-customer-for-user');
@@ -558,7 +558,7 @@ export class CustomerService {
             type: HistoryEntryType.CUSTOMER_PASSWORD_RESET_VERIFIED,
             data: {},
         });
-        this.eventBus.publish(new PasswordResetVerifiedEvent(ctx, result));
+        await this.eventBus.publish(new PasswordResetVerifiedEvent(ctx, result));
         return result;
     }
 
@@ -602,7 +602,7 @@ export class CustomerService {
         if (this.configService.authOptions.requireVerification) {
             user.getNativeAuthenticationMethod().pendingIdentifier = normalizedEmailAddress;
             await this.userService.setIdentifierChangeToken(ctx, user);
-            this.eventBus.publish(new IdentifierChangeRequestEvent(ctx, user));
+            await this.eventBus.publish(new IdentifierChangeRequestEvent(ctx, user));
             return true;
         } else {
             const oldIdentifier = user.identifier;
@@ -610,7 +610,7 @@ export class CustomerService {
             customer.emailAddress = normalizedEmailAddress;
             await this.connection.getRepository(ctx, User).save(user, { reload: false });
             await this.connection.getRepository(ctx, Customer).save(customer, { reload: false });
-            this.eventBus.publish(new IdentifierChangeEvent(ctx, user, oldIdentifier));
+            await this.eventBus.publish(new IdentifierChangeEvent(ctx, user, oldIdentifier));
             await this.historyService.createHistoryEntryForCustomer({
                 customerId: customer.id,
                 ctx,
@@ -645,7 +645,7 @@ export class CustomerService {
         if (!customer) {
             return false;
         }
-        this.eventBus.publish(new IdentifierChangeEvent(ctx, user, oldIdentifier));
+        await this.eventBus.publish(new IdentifierChangeEvent(ctx, user, oldIdentifier));
         customer.emailAddress = user.identifier;
         await this.connection.getRepository(ctx, Customer).save(customer, { reload: false });
         await this.historyService.createHistoryEntryForCustomer({
@@ -688,7 +688,7 @@ export class CustomerService {
         } else {
             customer = await this.connection.getRepository(ctx, Customer).save(new Customer(input));
             await this.channelService.assignToCurrentChannel(customer, ctx);
-            this.eventBus.publish(new CustomerEvent(ctx, customer, 'created', input));
+            await this.eventBus.publish(new CustomerEvent(ctx, customer, 'created', input));
         }
         return this.connection.getRepository(ctx, Customer).save(customer);
     }
@@ -721,7 +721,7 @@ export class CustomerService {
             data: { address: addressToLine(createdAddress) },
         });
         createdAddress.customer = customer;
-        this.eventBus.publish(new CustomerAddressEvent(ctx, createdAddress, 'created', input));
+        await this.eventBus.publish(new CustomerAddressEvent(ctx, createdAddress, 'created', input));
         return createdAddress;
     }
 
@@ -758,7 +758,7 @@ export class CustomerService {
             },
         });
         updatedAddress.customer = customer;
-        this.eventBus.publish(new CustomerAddressEvent(ctx, updatedAddress, 'updated', input));
+        await this.eventBus.publish(new CustomerAddressEvent(ctx, updatedAddress, 'updated', input));
         return updatedAddress;
     }
 
@@ -788,7 +788,7 @@ export class CustomerService {
         const deletedAddress = new Address(address);
         await this.connection.getRepository(ctx, Address).remove(address);
         address.customer = customer;
-        this.eventBus.publish(new CustomerAddressEvent(ctx, deletedAddress, 'deleted', id));
+        await this.eventBus.publish(new CustomerAddressEvent(ctx, deletedAddress, 'deleted', id));
         return true;
     }
 
@@ -803,7 +803,7 @@ export class CustomerService {
         if (customer.user) {
             await this.userService.softDelete(ctx, customer.user.id);
         }
-        this.eventBus.publish(new CustomerEvent(ctx, customer, 'deleted', customerId));
+        await this.eventBus.publish(new CustomerEvent(ctx, customer, 'deleted', customerId));
         return {
             result: DeletionResult.DELETED,
         };

packages/core/src/service/services/facet-value.service.ts

@@ -17,9 +17,9 @@ import { assertFound } from '../../common/utils';
 import { ConfigService } from '../../config/config.service';
 import { TransactionalConnection } from '../../connection/transactional-connection';
 import { Product, ProductVariant } from '../../entity';
+import { Facet } from '../../entity/facet/facet.entity';
 import { FacetValueTranslation } from '../../entity/facet-value/facet-value-translation.entity';
 import { FacetValue } from '../../entity/facet-value/facet-value.entity';
-import { Facet } from '../../entity/facet/facet.entity';
 import { EventBus } from '../../event-bus';
 import { FacetValueEvent } from '../../event-bus/events/facet-value-event';
 import { CustomFieldRelationService } from '../helpers/custom-field-relation/custom-field-relation.service';
@@ -185,7 +185,7 @@ export class FacetValueService {
             input as CreateFacetValueInput,
             facetValue,
         );
-        this.eventBus.publish(new FacetValueEvent(ctx, facetValueWithRelations, 'created', input));
+        await this.eventBus.publish(new FacetValueEvent(ctx, facetValueWithRelations, 'created', input));
         return assertFound(this.findOne(ctx, facetValue.id));
     }
 
@@ -197,7 +197,7 @@ export class FacetValueService {
             translationType: FacetValueTranslation,
         });
         await this.customFieldRelationService.updateRelations(ctx, FacetValue, input, facetValue);
-        this.eventBus.publish(new FacetValueEvent(ctx, facetValue, 'updated', input));
+        await this.eventBus.publish(new FacetValueEvent(ctx, facetValue, 'updated', input));
         return assertFound(this.findOne(ctx, facetValue.id));
     }
 
@@ -222,11 +222,11 @@ export class FacetValueService {
 
         if (!isInUse) {
             await this.connection.getRepository(ctx, FacetValue).remove(facetValue);
-            this.eventBus.publish(new FacetValueEvent(ctx, deletedFacetValue, 'deleted', id));
+            await this.eventBus.publish(new FacetValueEvent(ctx, deletedFacetValue, 'deleted', id));
             result = DeletionResult.DELETED;
         } else if (force) {
             await this.connection.getRepository(ctx, FacetValue).remove(facetValue);
-            this.eventBus.publish(new FacetValueEvent(ctx, deletedFacetValue, 'deleted', id));
+            await this.eventBus.publish(new FacetValueEvent(ctx, deletedFacetValue, 'deleted', id));
             message = ctx.translate('message.facet-value-force-deleted', i18nVars);
             result = DeletionResult.DELETED;
         } else {

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

@@ -170,7 +170,7 @@ export class FacetService {
             input,
             facet,
         );
-        this.eventBus.publish(new FacetEvent(ctx, facetWithRelations, 'created', input));
+        await this.eventBus.publish(new FacetEvent(ctx, facetWithRelations, 'created', input));
         return assertFound(this.findOne(ctx, facet.id));
     }
 
@@ -187,7 +187,7 @@ export class FacetService {
             },
         });
         await this.customFieldRelationService.updateRelations(ctx, Facet, input, facet);
-        this.eventBus.publish(new FacetEvent(ctx, facet, 'updated', input));
+        await this.eventBus.publish(new FacetEvent(ctx, facet, 'updated', input));
         return assertFound(this.findOne(ctx, facet.id));
     }
 
@@ -216,11 +216,11 @@ export class FacetService {
 
         if (!isInUse) {
             await this.connection.getRepository(ctx, Facet).remove(facet);
-            this.eventBus.publish(new FacetEvent(ctx, deletedFacet, 'deleted', id));
+            await this.eventBus.publish(new FacetEvent(ctx, deletedFacet, 'deleted', id));
             result = DeletionResult.DELETED;
         } else if (force) {
             await this.connection.getRepository(ctx, Facet).remove(facet);
-            this.eventBus.publish(new FacetEvent(ctx, deletedFacet, 'deleted', id));
+            await this.eventBus.publish(new FacetEvent(ctx, deletedFacet, 'deleted', id));
             message = ctx.translate('message.facet-force-deleted', i18nVars);
             result = DeletionResult.DELETED;
         } else {

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

@@ -110,7 +110,7 @@ export class FulfillmentService {
             fulfillmentPartial,
             newFulfillment,
         );
-        this.eventBus.publish(
+        await this.eventBus.publish(
             new FulfillmentEvent(ctx, fulfillmentWithRelations, {
                 orders,
                 lines,
@@ -183,7 +183,7 @@ export class FulfillmentService {
             return new FulfillmentStateTransitionError({ transitionError, fromState, toState: state });
         }
         await this.connection.getRepository(ctx, Fulfillment).save(fulfillment, { reload: false });
-        this.eventBus.publish(new FulfillmentStateTransitionEvent(fromState, state, ctx, fulfillment));
+        await this.eventBus.publish(new FulfillmentStateTransitionEvent(fromState, state, ctx, fulfillment));
         await finalize();
         return { fulfillment, orders, fromState, toState: state };
     }

packages/core/src/service/services/global-settings.service.ts

@@ -73,7 +73,7 @@ export class GlobalSettingsService {
 
     async updateSettings(ctx: RequestContext, input: UpdateGlobalSettingsInput): Promise<GlobalSettings> {
         const settings = await this.getSettings(ctx);
-        this.eventBus.publish(new GlobalSettingsEvent(ctx, settings, input));
+        await this.eventBus.publish(new GlobalSettingsEvent(ctx, settings, input));
         patchEntity(settings, input);
         await this.customFieldRelationService.updateRelations(ctx, GlobalSettings, input, settings);
         return this.connection.getRepository(ctx, GlobalSettings).save(settings);

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

@@ -290,7 +290,7 @@ export class HistoryService {
             administrator,
         });
         const history = await this.connection.getRepository(ctx, OrderHistoryEntry).save(entry);
-        this.eventBus.publish(new HistoryEntryEvent(ctx, history, 'created', 'order', { type, data }));
+        await this.eventBus.publish(new HistoryEntryEvent(ctx, history, 'created', 'order', { type, data }));
         return history;
     }
 
@@ -331,7 +331,9 @@ export class HistoryService {
             administrator,
         });
         const history = await this.connection.getRepository(ctx, CustomerHistoryEntry).save(entry);
-        this.eventBus.publish(new HistoryEntryEvent(ctx, history, 'created', 'customer', { type, data }));
+        await this.eventBus.publish(
+            new HistoryEntryEvent(ctx, history, 'created', 'customer', { type, data }),
+        );
         return history;
     }
 
@@ -354,7 +356,7 @@ export class HistoryService {
             entry.administrator = administrator;
         }
         const newEntry = await this.connection.getRepository(ctx, OrderHistoryEntry).save(entry);
-        this.eventBus.publish(new HistoryEntryEvent(ctx, entry, 'updated', 'order', args));
+        await this.eventBus.publish(new HistoryEntryEvent(ctx, entry, 'updated', 'order', args));
         return newEntry;
     }
 
@@ -362,7 +364,7 @@ export class HistoryService {
         const entry = await this.connection.getEntityOrThrow(ctx, OrderHistoryEntry, id);
         const deletedEntry = new OrderHistoryEntry(entry);
         await this.connection.getRepository(ctx, OrderHistoryEntry).remove(entry);
-        this.eventBus.publish(new HistoryEntryEvent(ctx, deletedEntry, 'deleted', 'order', id));
+        await this.eventBus.publish(new HistoryEntryEvent(ctx, deletedEntry, 'deleted', 'order', id));
     }
 
     async updateCustomerHistoryEntry<T extends keyof CustomerHistoryEntryData>(
@@ -381,7 +383,7 @@ export class HistoryService {
             entry.administrator = administrator;
         }
         const newEntry = await this.connection.getRepository(ctx, CustomerHistoryEntry).save(entry);
-        this.eventBus.publish(new HistoryEntryEvent(ctx, entry, 'updated', 'customer', args));
+        await this.eventBus.publish(new HistoryEntryEvent(ctx, entry, 'updated', 'customer', args));
         return newEntry;
     }
 
@@ -389,7 +391,7 @@ export class HistoryService {
         const entry = await this.connection.getEntityOrThrow(ctx, CustomerHistoryEntry, id);
         const deletedEntry = new CustomerHistoryEntry(entry);
         await this.connection.getRepository(ctx, CustomerHistoryEntry).remove(entry);
-        this.eventBus.publish(new HistoryEntryEvent(ctx, deletedEntry, 'deleted', 'customer', id));
+        await this.eventBus.publish(new HistoryEntryEvent(ctx, deletedEntry, 'deleted', 'customer', id));
     }
 
     private async getAdministratorFromContext(ctx: RequestContext): Promise<Administrator | undefined> {

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

@@ -424,7 +424,7 @@ export class OrderService {
         }
         await this.channelService.assignToCurrentChannel(newOrder, ctx);
         const order = await this.connection.getRepository(ctx, Order).save(newOrder);
-        this.eventBus.publish(new OrderEvent(ctx, order, 'created'));
+        await this.eventBus.publish(new OrderEvent(ctx, order, 'created'));
         const transitionResult = await this.transitionToState(ctx, order.id, 'AddingItems');
         if (isGraphQlErrorResult(transitionResult)) {
             // this should never occur, so we will throw rather than return
@@ -438,7 +438,7 @@ export class OrderService {
         newOrder.active = false;
         await this.channelService.assignToCurrentChannel(newOrder, ctx);
         const order = await this.connection.getRepository(ctx, Order).save(newOrder);
-        this.eventBus.publish(new OrderEvent(ctx, order, 'created'));
+        await this.eventBus.publish(new OrderEvent(ctx, order, 'created'));
         const transitionResult = await this.transitionToState(ctx, order.id, 'Draft');
         if (isGraphQlErrorResult(transitionResult)) {
             // this should never occur, so we will throw rather than return
@@ -473,7 +473,7 @@ export class OrderService {
         order = patchEntity(order, { customFields });
         await this.customFieldRelationService.updateRelations(ctx, Order, { customFields }, order);
         const updatedOrder = await this.connection.getRepository(ctx, Order).save(order);
-        this.eventBus.publish(new OrderEvent(ctx, updatedOrder, 'updated'));
+        await this.eventBus.publish(new OrderEvent(ctx, updatedOrder, 'updated'));
         return updatedOrder;
     }
 
@@ -507,7 +507,7 @@ export class OrderService {
         }
 
         const updatedOrder = await this.addCustomerToOrder(ctx, order.id, targetCustomer);
-        this.eventBus.publish(new OrderEvent(ctx, updatedOrder, 'updated'));
+        await this.eventBus.publish(new OrderEvent(ctx, updatedOrder, 'updated'));
         await this.historyService.createHistoryEntryForOrder({
             ctx,
             orderId,
@@ -650,7 +650,7 @@ export class OrderService {
             order.lines = order.lines.filter(l => !idsAreEqual(l.id, orderLine.id));
             const deletedOrderLine = new OrderLine(orderLine);
             await this.connection.getRepository(ctx, OrderLine).remove(orderLine);
-            this.eventBus.publish(new OrderLineEvent(ctx, order, deletedOrderLine, 'deleted'));
+            await this.eventBus.publish(new OrderLineEvent(ctx, order, deletedOrderLine, 'deleted'));
             updatedOrderLines = [];
         } else {
             await this.orderModifier.updateOrderLineQuantity(ctx, orderLine, correctedQuantity, order);
@@ -688,7 +688,7 @@ export class OrderService {
         const updatedOrder = await this.applyPriceAdjustments(ctx, order);
         const deletedOrderLine = new OrderLine(orderLine);
         await this.connection.getRepository(ctx, OrderLine).remove(orderLine);
-        this.eventBus.publish(new OrderLineEvent(ctx, order, deletedOrderLine, 'deleted'));
+        await this.eventBus.publish(new OrderLineEvent(ctx, order, deletedOrderLine, 'deleted'));
         return updatedOrder;
     }
 
@@ -781,7 +781,7 @@ export class OrderService {
             type: HistoryEntryType.ORDER_COUPON_APPLIED,
             data: { couponCode, promotionId: validationResult.id },
         });
-        this.eventBus.publish(new CouponCodeEvent(ctx, couponCode, orderId, 'assigned'));
+        await this.eventBus.publish(new CouponCodeEvent(ctx, couponCode, orderId, 'assigned'));
         return this.applyPriceAdjustments(ctx, order);
     }
 
@@ -807,7 +807,7 @@ export class OrderService {
                 type: HistoryEntryType.ORDER_COUPON_REMOVED,
                 data: { couponCode },
             });
-            this.eventBus.publish(new CouponCodeEvent(ctx, couponCode, orderId, 'removed'));
+            await this.eventBus.publish(new CouponCodeEvent(ctx, couponCode, orderId, 'removed'));
             const result = await this.applyPriceAdjustments(ctx, order);
             await this.connection.getRepository(ctx, OrderLine).save(affectedOrderLines);
             return result;
@@ -961,7 +961,7 @@ export class OrderService {
             return new OrderStateTransitionError({ transitionError, fromState, toState: state });
         }
         await this.connection.getRepository(ctx, Order).save(order, { reload: false });
-        this.eventBus.publish(new OrderStateTransitionEvent(fromState, state, ctx, order));
+        await this.eventBus.publish(new OrderStateTransitionEvent(fromState, state, ctx, order));
         await finalize();
         await this.connection.getRepository(ctx, Order).save(order, { reload: false });
         return order;
@@ -1438,7 +1438,7 @@ export class OrderService {
         );
         await this.connection.getRepository(ctx, Refund).save(refund);
         await finalize();
-        this.eventBus.publish(
+        await this.eventBus.publish(
             new RefundStateTransitionEvent(fromState, toState, ctx, refund, refund.payment.order),
         );
         return refund;

packages/core/src/service/services/payment-method.service.ts

@@ -109,7 +109,7 @@ export class PaymentMethodService {
             },
         });
         await this.customFieldRelationService.updateRelations(ctx, PaymentMethod, input, savedPaymentMethod);
-        this.eventBus.publish(new PaymentMethodEvent(ctx, savedPaymentMethod, 'created', input));
+        await this.eventBus.publish(new PaymentMethodEvent(ctx, savedPaymentMethod, 'created', input));
         return assertFound(this.findOne(ctx, savedPaymentMethod.id));
     }
 
@@ -140,7 +140,7 @@ export class PaymentMethodService {
             input,
             updatedPaymentMethod,
         );
-        this.eventBus.publish(new PaymentMethodEvent(ctx, updatedPaymentMethod, 'updated', input));
+        await this.eventBus.publish(new PaymentMethodEvent(ctx, updatedPaymentMethod, 'updated', input));
         await this.connection.getRepository(ctx, PaymentMethod).save(updatedPaymentMethod, { reload: false });
         return assertFound(this.findOne(ctx, updatedPaymentMethod.id));
     }
@@ -168,7 +168,7 @@ export class PaymentMethodService {
             try {
                 const deletedPaymentMethod = new PaymentMethod(paymentMethod);
                 await this.connection.getRepository(ctx, PaymentMethod).remove(paymentMethod);
-                this.eventBus.publish(
+                await this.eventBus.publish(
                     new PaymentMethodEvent(ctx, deletedPaymentMethod, 'deleted', paymentMethodId),
                 );
                 return {
@@ -185,7 +185,9 @@ export class PaymentMethodService {
             // but will remove from the current channel
             paymentMethod.channels = paymentMethod.channels.filter(c => !idsAreEqual(c.id, ctx.channelId));
             await this.connection.getRepository(ctx, PaymentMethod).save(paymentMethod);
-            this.eventBus.publish(new PaymentMethodEvent(ctx, paymentMethod, 'deleted', paymentMethodId));
+            await this.eventBus.publish(
+                new PaymentMethodEvent(ctx, paymentMethod, 'deleted', paymentMethodId),
+            );
             return {
                 result: DeletionResult.DELETED,
             };

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

@@ -139,7 +139,7 @@ export class PaymentService {
             .relation('payments')
             .of(order)
             .add(payment);
-        this.eventBus.publish(
+        await this.eventBus.publish(
             new PaymentStateTransitionEvent(initialState, result.state, ctx, payment, order),
         );
         await finalize();
@@ -228,7 +228,7 @@ export class PaymentService {
             return new PaymentStateTransitionError({ transitionError, fromState, toState });
         }
         await this.connection.getRepository(ctx, Payment).save(payment, { reload: false });
-        this.eventBus.publish(
+        await this.eventBus.publish(
             new PaymentStateTransitionEvent(fromState, toState, ctx, payment, payment.order),
         );
         await finalize();
@@ -264,7 +264,9 @@ export class PaymentService {
             .relation('payments')
             .of(order)
             .add(payment);
-        this.eventBus.publish(new PaymentStateTransitionEvent(initialState, endState, ctx, payment, order));
+        await this.eventBus.publish(
+            new PaymentStateTransitionEvent(initialState, endState, ctx, payment, order),
+        );
         await finalize();
         return payment;
     }
@@ -404,7 +406,7 @@ export class PaymentService {
                 }
                 await this.connection.getRepository(ctx, Refund).save(refund, { reload: false });
                 await finalize();
-                this.eventBus.publish(
+                await this.eventBus.publish(
                     new RefundStateTransitionEvent(fromState, createRefundResult.state, ctx, refund, order),
                 );
             }

packages/core/src/service/services/product-option-group.service.ts

@@ -106,7 +106,7 @@ export class ProductOptionGroupService {
             input,
             group,
         );
-        this.eventBus.publish(new ProductOptionGroupEvent(ctx, groupWithRelations, 'created', input));
+        await this.eventBus.publish(new ProductOptionGroupEvent(ctx, groupWithRelations, 'created', input));
         return assertFound(this.findOne(ctx, group.id));
     }
 
@@ -121,7 +121,7 @@ export class ProductOptionGroupService {
             translationType: ProductOptionGroupTranslation,
         });
         await this.customFieldRelationService.updateRelations(ctx, ProductOptionGroup, input, group);
-        this.eventBus.publish(new ProductOptionGroupEvent(ctx, group, 'updated', input));
+        await this.eventBus.publish(new ProductOptionGroupEvent(ctx, group, 'updated', input));
         return assertFound(this.findOne(ctx, group.id));
     }
 
@@ -183,7 +183,7 @@ export class ProductOptionGroupService {
                 Logger.error(e.message, undefined, e.stack);
             }
         }
-        this.eventBus.publish(new ProductOptionGroupEvent(ctx, deletedOptionGroup, 'deleted', id));
+        await this.eventBus.publish(new ProductOptionGroupEvent(ctx, deletedOptionGroup, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
         };

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

@@ -80,7 +80,7 @@ export class ProductOptionService {
             input as CreateProductOptionInput,
             option,
         );
-        this.eventBus.publish(new ProductOptionEvent(ctx, optionWithRelations, 'created', input));
+        await this.eventBus.publish(new ProductOptionEvent(ctx, optionWithRelations, 'created', input));
         return assertFound(this.findOne(ctx, option.id));
     }
 
@@ -92,7 +92,7 @@ export class ProductOptionService {
             translationType: ProductOptionTranslation,
         });
         await this.customFieldRelationService.updateRelations(ctx, ProductOption, input, option);
-        this.eventBus.publish(new ProductOptionEvent(ctx, option, 'updated', input));
+        await this.eventBus.publish(new ProductOptionEvent(ctx, option, 'updated', input));
         return assertFound(this.findOne(ctx, option.id));
     }
 
@@ -131,7 +131,7 @@ export class ProductOptionService {
                 Logger.error(e.message, undefined, e.stack);
             }
         }
-        this.eventBus.publish(new ProductOptionEvent(ctx, deletedProductOption, 'deleted', id));
+        await this.eventBus.publish(new ProductOptionEvent(ctx, deletedProductOption, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
         };

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

@@ -386,7 +386,7 @@ export class ProductVariantService {
             ids.push(id);
         }
         const createdVariants = await this.findByIds(ctx, ids);
-        this.eventBus.publish(new ProductVariantEvent(ctx, createdVariants, 'created', input));
+        await this.eventBus.publish(new ProductVariantEvent(ctx, createdVariants, 'created', input));
         return createdVariants;
     }
 
@@ -401,7 +401,7 @@ export class ProductVariantService {
             ctx,
             input.map(i => i.id),
         );
-        this.eventBus.publish(new ProductVariantEvent(ctx, updatedVariants, 'updated', input));
+        await this.eventBus.publish(new ProductVariantEvent(ctx, updatedVariants, 'updated', input));
         return updatedVariants;
     }
 
@@ -600,7 +600,7 @@ export class ProductVariantService {
                     currencyCode: currencyCode ?? ctx.channel.defaultCurrencyCode,
                 }),
             );
-            this.eventBus.publish(new ProductVariantPriceEvent(ctx, [createdPrice], 'created'));
+            await this.eventBus.publish(new ProductVariantPriceEvent(ctx, [createdPrice], 'created'));
             additionalPricesToUpdate = await productVariantPriceUpdateStrategy.onPriceCreated(
                 ctx,
                 createdPrice,
@@ -612,7 +612,7 @@ export class ProductVariantService {
             const updatedPrice = await this.connection
                 .getRepository(ctx, ProductVariantPrice)
                 .save(targetPrice);
-            this.eventBus.publish(new ProductVariantPriceEvent(ctx, [updatedPrice], 'updated'));
+            await this.eventBus.publish(new ProductVariantPriceEvent(ctx, [updatedPrice], 'updated'));
             additionalPricesToUpdate = await productVariantPriceUpdateStrategy.onPriceUpdated(
                 ctx,
                 updatedPrice,
@@ -630,7 +630,9 @@ export class ProductVariantService {
             const updatedAdditionalPrices = await this.connection
                 .getRepository(ctx, ProductVariantPrice)
                 .save(uniqueAdditionalPricesToUpdate);
-            this.eventBus.publish(new ProductVariantPriceEvent(ctx, updatedAdditionalPrices, 'updated'));
+            await this.eventBus.publish(
+                new ProductVariantPriceEvent(ctx, updatedAdditionalPrices, 'updated'),
+            );
         }
         return targetPrice;
     }
@@ -650,7 +652,7 @@ export class ProductVariantService {
         });
         if (variantPrice) {
             await this.connection.getRepository(ctx, ProductVariantPrice).remove(variantPrice);
-            this.eventBus.publish(new ProductVariantPriceEvent(ctx, [variantPrice], 'deleted'));
+            await this.eventBus.publish(new ProductVariantPriceEvent(ctx, [variantPrice], 'deleted'));
             const { productVariantPriceUpdateStrategy } = this.configService.catalogOptions;
             const allPrices = await this.connection.getRepository(ctx, ProductVariantPrice).find({
                 where: {
@@ -666,7 +668,9 @@ export class ProductVariantService {
                 const updatedAdditionalPrices = await this.connection
                     .getRepository(ctx, ProductVariantPrice)
                     .save(additionalPricesToUpdate);
-                this.eventBus.publish(new ProductVariantPriceEvent(ctx, updatedAdditionalPrices, 'updated'));
+                await this.eventBus.publish(
+                    new ProductVariantPriceEvent(ctx, updatedAdditionalPrices, 'updated'),
+                );
             }
         }
     }
@@ -680,7 +684,7 @@ export class ProductVariantService {
             variant.deletedAt = new Date();
         }
         await this.connection.getRepository(ctx, ProductVariant).save(variants, { reload: false });
-        this.eventBus.publish(new ProductVariantEvent(ctx, variants, 'deleted', id));
+        await this.eventBus.publish(new ProductVariantEvent(ctx, variants, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
         };
@@ -817,7 +821,9 @@ export class ProductVariantService {
             variants.map(v => v.id),
         );
         for (const variant of variants) {
-            this.eventBus.publish(new ProductVariantChannelEvent(ctx, variant, input.channelId, 'assigned'));
+            await this.eventBus.publish(
+                new ProductVariantChannelEvent(ctx, variant, input.channelId, 'assigned'),
+            );
         }
         return result;
     }
@@ -871,7 +877,9 @@ export class ProductVariantService {
         // whereby an event listener triggers a query which does not yet have access to the changes
         // within the current transaction.
         for (const variant of variants) {
-            this.eventBus.publish(new ProductVariantChannelEvent(ctx, variant, input.channelId, 'removed'));
+            await this.eventBus.publish(
+                new ProductVariantChannelEvent(ctx, variant, input.channelId, 'removed'),
+            );
         }
         return result;
     }

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

@@ -235,7 +235,7 @@ export class ProductService {
         });
         await this.customFieldRelationService.updateRelations(ctx, Product, input, product);
         await this.assetService.updateEntityAssets(ctx, product, input);
-        this.eventBus.publish(new ProductEvent(ctx, product, 'created', input));
+        await this.eventBus.publish(new ProductEvent(ctx, product, 'created', input));
         return assertFound(this.findOne(ctx, product.id));
     }
 
@@ -265,7 +265,7 @@ export class ProductService {
             },
         });
         await this.customFieldRelationService.updateRelations(ctx, Product, input, updatedProduct);
-        this.eventBus.publish(new ProductEvent(ctx, updatedProduct, 'updated', input));
+        await this.eventBus.publish(new ProductEvent(ctx, updatedProduct, 'updated', input));
         return assertFound(this.findOne(ctx, updatedProduct.id));
     }
 
@@ -278,7 +278,7 @@ export class ProductService {
         });
         product.deletedAt = new Date();
         await this.connection.getRepository(ctx, Product).save(product, { reload: false });
-        this.eventBus.publish(new ProductEvent(ctx, product, 'deleted', productId));
+        await this.eventBus.publish(new ProductEvent(ctx, product, 'deleted', productId));
 
         const variantResult = await this.productVariantService.softDelete(
             ctx,
@@ -335,7 +335,7 @@ export class ProductService {
             .getRepository(ctx, Product)
             .find({ where: { id: In(input.productIds) } });
         for (const product of products) {
-            this.eventBus.publish(new ProductChannelEvent(ctx, product, input.channelId, 'assigned'));
+            await this.eventBus.publish(new ProductChannelEvent(ctx, product, input.channelId, 'assigned'));
         }
         return this.findByIds(
             ctx,
@@ -361,7 +361,7 @@ export class ProductService {
             .getRepository(ctx, Product)
             .find({ where: { id: In(input.productIds) } });
         for (const product of products) {
-            this.eventBus.publish(new ProductChannelEvent(ctx, product, input.channelId, 'removed'));
+            await this.eventBus.publish(new ProductChannelEvent(ctx, product, input.channelId, 'removed'));
         }
         return this.findByIds(
             ctx,
@@ -397,7 +397,9 @@ export class ProductService {
         }
 
         await this.connection.getRepository(ctx, Product).save(product, { reload: false });
-        this.eventBus.publish(new ProductOptionGroupChangeEvent(ctx, product, optionGroupId, 'assigned'));
+        await this.eventBus.publish(
+            new ProductOptionGroupChangeEvent(ctx, product, optionGroupId, 'assigned'),
+        );
         return assertFound(this.findOne(ctx, productId));
     }
 
@@ -445,7 +447,9 @@ export class ProductService {
             // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
             throw new InternalServerError(result.message!);
         }
-        this.eventBus.publish(new ProductOptionGroupChangeEvent(ctx, product, optionGroupId, 'removed'));
+        await this.eventBus.publish(
+            new ProductOptionGroupChangeEvent(ctx, product, optionGroupId, 'removed'),
+        );
         return assertFound(this.findOne(ctx, productId));
     }
 

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

@@ -148,7 +148,7 @@ export class PromotionService {
             input,
             newPromotion,
         );
-        this.eventBus.publish(new PromotionEvent(ctx, promotionWithRelations, 'created', input));
+        await this.eventBus.publish(new PromotionEvent(ctx, promotionWithRelations, 'created', input));
         return assertFound(this.findOne(ctx, newPromotion.id));
     }
 
@@ -187,7 +187,7 @@ export class PromotionService {
             },
         });
         await this.customFieldRelationService.updateRelations(ctx, Promotion, input, updatedPromotion);
-        this.eventBus.publish(new PromotionEvent(ctx, promotion, 'updated', input));
+        await this.eventBus.publish(new PromotionEvent(ctx, promotion, 'updated', input));
         return assertFound(this.findOne(ctx, updatedPromotion.id));
     }
 
@@ -196,7 +196,7 @@ export class PromotionService {
         await this.connection
             .getRepository(ctx, Promotion)
             .update({ id: promotionId }, { deletedAt: new Date() });
-        this.eventBus.publish(new PromotionEvent(ctx, promotion, 'deleted', promotionId));
+        await this.eventBus.publish(new PromotionEvent(ctx, promotion, 'deleted', promotionId));
 
         return {
             result: DeletionResult.DELETED,

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

@@ -73,7 +73,7 @@ export class ProvinceService {
             entityType: Province as Type<Region>,
             translationType: RegionTranslation,
         });
-        this.eventBus.publish(new ProvinceEvent(ctx, province, 'created', input));
+        await this.eventBus.publish(new ProvinceEvent(ctx, province, 'created', input));
         return assertFound(this.findOne(ctx, province.id));
     }
 
@@ -84,7 +84,7 @@ export class ProvinceService {
             entityType: Province as Type<Region>,
             translationType: RegionTranslation,
         });
-        this.eventBus.publish(new ProvinceEvent(ctx, province, 'updated', input));
+        await this.eventBus.publish(new ProvinceEvent(ctx, province, 'updated', input));
         return assertFound(this.findOne(ctx, province.id));
     }
 
@@ -93,7 +93,7 @@ export class ProvinceService {
 
         const deletedProvince = new Province(region);
         await this.connection.getRepository(ctx, Province).remove(region);
-        this.eventBus.publish(new ProvinceEvent(ctx, deletedProvince, 'deleted', id));
+        await this.eventBus.publish(new ProvinceEvent(ctx, deletedProvince, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
             message: '',

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

@@ -231,7 +231,7 @@ export class RoleService {
         }
         await this.checkActiveUserHasSufficientPermissions(ctx, targetChannels, input.permissions);
         const role = await this.createRoleForChannels(ctx, input, targetChannels);
-        this.eventBus.publish(new RoleEvent(ctx, role, 'created', input));
+        await this.eventBus.publish(new RoleEvent(ctx, role, 'created', input));
         return role;
     }
 
@@ -265,7 +265,7 @@ export class RoleService {
             updatedRole.channels = targetChannels;
         }
         await this.connection.getRepository(ctx, Role).save(updatedRole, { reload: false });
-        this.eventBus.publish(new RoleEvent(ctx, role, 'updated', input));
+        await this.eventBus.publish(new RoleEvent(ctx, role, 'updated', input));
         return await assertFound(this.findOne(ctx, role.id));
     }
 
@@ -279,7 +279,7 @@ export class RoleService {
         }
         const deletedRole = new Role(role);
         await this.connection.getRepository(ctx, Role).remove(role);
-        this.eventBus.publish(new RoleEvent(ctx, deletedRole, 'deleted', id));
+        await this.eventBus.publish(new RoleEvent(ctx, deletedRole, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
         };

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

@@ -61,7 +61,7 @@ export class SellerService {
             input,
             seller,
         );
-        this.eventBus.publish(new SellerEvent(ctx, sellerWithRelations, 'created', input));
+        await this.eventBus.publish(new SellerEvent(ctx, sellerWithRelations, 'created', input));
         return assertFound(this.findOne(ctx, seller.id));
     }
 
@@ -75,7 +75,7 @@ export class SellerService {
             input,
             seller,
         );
-        this.eventBus.publish(new SellerEvent(ctx, sellerWithRelations, 'updated', input));
+        await this.eventBus.publish(new SellerEvent(ctx, sellerWithRelations, 'updated', input));
         return seller;
     }
 
@@ -83,7 +83,7 @@ export class SellerService {
         const seller = await this.connection.getEntityOrThrow(ctx, Seller, id);
         await this.connection.getRepository(ctx, Seller).remove(seller);
         const deletedSeller = new Seller(seller);
-        this.eventBus.publish(new SellerEvent(ctx, deletedSeller, 'deleted', id));
+        await this.eventBus.publish(new SellerEvent(ctx, deletedSeller, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
         };

packages/core/src/service/services/shipping-method.service.ts

@@ -133,7 +133,9 @@ export class ShippingMethodService {
             input,
             newShippingMethod,
         );
-        this.eventBus.publish(new ShippingMethodEvent(ctx, shippingMethodWithRelations, 'created', input));
+        await this.eventBus.publish(
+            new ShippingMethodEvent(ctx, shippingMethodWithRelations, 'created', input),
+        );
         return assertFound(this.findOne(ctx, newShippingMethod.id));
     }
 
@@ -175,7 +177,7 @@ export class ShippingMethodService {
             input,
             updatedShippingMethod,
         );
-        this.eventBus.publish(new ShippingMethodEvent(ctx, shippingMethod, 'updated', input));
+        await this.eventBus.publish(new ShippingMethodEvent(ctx, shippingMethod, 'updated', input));
         return assertFound(this.findOne(ctx, shippingMethod.id));
     }
 
@@ -186,7 +188,7 @@ export class ShippingMethodService {
         });
         shippingMethod.deletedAt = new Date();
         await this.connection.getRepository(ctx, ShippingMethod).save(shippingMethod, { reload: false });
-        this.eventBus.publish(new ShippingMethodEvent(ctx, shippingMethod, 'deleted', id));
+        await this.eventBus.publish(new ShippingMethodEvent(ctx, shippingMethod, 'deleted', id));
         return {
             result: DeletionResult.DELETED,
         };

packages/core/src/service/services/stock-movement.service.ts

@@ -14,8 +14,8 @@ import { ConfigService } from '../../config/config.service';
 import { ShippingCalculator } from '../../config/shipping-method/shipping-calculator';
 import { ShippingEligibilityChecker } from '../../config/shipping-method/shipping-eligibility-checker';
 import { TransactionalConnection } from '../../connection/transactional-connection';
-import { OrderLine } from '../../entity/order-line/order-line.entity';
 import { Order } from '../../entity/order/order.entity';
+import { OrderLine } from '../../entity/order-line/order-line.entity';
 import { ProductVariant } from '../../entity/product-variant/product-variant.entity';
 import { ShippingMethod } from '../../entity/shipping-method/shipping-method.entity';
 import { Allocation } from '../../entity/stock-movement/allocation.entity';
@@ -121,7 +121,7 @@ export class StockMovementService {
                 input.stockLocationId,
                 delta,
             );
-            this.eventBus.publish(new StockMovementEvent(ctx, [adjustment]));
+            await this.eventBus.publish(new StockMovementEvent(ctx, [adjustment]));
             adjustments.push(adjustment);
         }
 
@@ -187,7 +187,7 @@ export class StockMovementService {
         }
         const savedAllocations = await this.connection.getRepository(ctx, Allocation).save(allocations);
         if (savedAllocations.length) {
-            this.eventBus.publish(new StockMovementEvent(ctx, savedAllocations));
+            await this.eventBus.publish(new StockMovementEvent(ctx, savedAllocations));
         }
         return savedAllocations;
     }
@@ -247,7 +247,7 @@ export class StockMovementService {
         }
         const savedSales = await this.connection.getRepository(ctx, Sale).save(sales);
         if (savedSales.length) {
-            this.eventBus.publish(new StockMovementEvent(ctx, savedSales));
+            await this.eventBus.publish(new StockMovementEvent(ctx, savedSales));
         }
         return savedSales;
     }
@@ -302,7 +302,7 @@ export class StockMovementService {
         }
         const savedCancellations = await this.connection.getRepository(ctx, Cancellation).save(cancellations);
         if (savedCancellations.length) {
-            this.eventBus.publish(new StockMovementEvent(ctx, savedCancellations));
+            await this.eventBus.publish(new StockMovementEvent(ctx, savedCancellations));
         }
         return savedCancellations;
     }
@@ -351,7 +351,7 @@ export class StockMovementService {
         }
         const savedReleases = await this.connection.getRepository(ctx, Release).save(releases);
         if (savedReleases.length) {
-            this.eventBus.publish(new StockMovementEvent(ctx, savedReleases));
+            await this.eventBus.publish(new StockMovementEvent(ctx, savedReleases));
         }
         return savedReleases;
     }

packages/core/src/service/services/tax-category.service.ts

@@ -61,7 +61,7 @@ export class TaxCategoryService {
                 .update({ isDefault: true }, { isDefault: false });
         }
         const newTaxCategory = await this.connection.getRepository(ctx, TaxCategory).save(taxCategory);
-        this.eventBus.publish(new TaxCategoryEvent(ctx, newTaxCategory, 'created', input));
+        await this.eventBus.publish(new TaxCategoryEvent(ctx, newTaxCategory, 'created', input));
         return assertFound(this.findOne(ctx, newTaxCategory.id));
     }
 
@@ -77,7 +77,7 @@ export class TaxCategoryService {
                 .update({ isDefault: true }, { isDefault: false });
         }
         await this.connection.getRepository(ctx, TaxCategory).save(updatedTaxCategory, { reload: false });
-        this.eventBus.publish(new TaxCategoryEvent(ctx, taxCategory, 'updated', input));
+        await this.eventBus.publish(new TaxCategoryEvent(ctx, taxCategory, 'updated', input));
         return assertFound(this.findOne(ctx, taxCategory.id));
     }
 
@@ -101,7 +101,7 @@ export class TaxCategoryService {
         try {
             const deletedTaxCategory = new TaxCategory(taxCategory);
             await this.connection.getRepository(ctx, TaxCategory).remove(taxCategory);
-            this.eventBus.publish(new TaxCategoryEvent(ctx, deletedTaxCategory, 'deleted', id));
+            await this.eventBus.publish(new TaxCategoryEvent(ctx, deletedTaxCategory, 'deleted', id));
             return {
                 result: DeletionResult.DELETED,
             };

packages/core/src/service/services/tax-rate.service.ts

@@ -97,8 +97,8 @@ export class TaxRateService {
         }
         const newTaxRate = await this.connection.getRepository(ctx, TaxRate).save(taxRate);
         await this.updateActiveTaxRates(ctx);
-        this.eventBus.publish(new TaxRateModificationEvent(ctx, newTaxRate));
-        this.eventBus.publish(new TaxRateEvent(ctx, newTaxRate, 'created', input));
+        await this.eventBus.publish(new TaxRateModificationEvent(ctx, newTaxRate));
+        await this.eventBus.publish(new TaxRateEvent(ctx, newTaxRate, 'created', input));
         return assertFound(this.findOne(ctx, newTaxRate.id));
     }
 
@@ -132,8 +132,8 @@ export class TaxRateService {
         // TaxRate when updating its own tax rate cache.
         await this.connection.commitOpenTransaction(ctx);
 
-        this.eventBus.publish(new TaxRateModificationEvent(ctx, updatedTaxRate));
-        this.eventBus.publish(new TaxRateEvent(ctx, updatedTaxRate, 'updated', input));
+        await this.eventBus.publish(new TaxRateModificationEvent(ctx, updatedTaxRate));
+        await this.eventBus.publish(new TaxRateEvent(ctx, updatedTaxRate, 'updated', input));
 
         return assertFound(this.findOne(ctx, taxRate.id));
     }
@@ -143,7 +143,7 @@ export class TaxRateService {
         const deletedTaxRate = new TaxRate(taxRate);
         try {
             await this.connection.getRepository(ctx, TaxRate).remove(taxRate);
-            this.eventBus.publish(new TaxRateEvent(ctx, deletedTaxRate, 'deleted', id));
+            await this.eventBus.publish(new TaxRateEvent(ctx, deletedTaxRate, 'deleted', id));
             return {
                 result: DeletionResult.DELETED,
             };

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

@@ -120,7 +120,7 @@ export class ZoneService {
         }
         const newZone = await this.connection.getRepository(ctx, Zone).save(zone);
         await this.zones.refresh(ctx);
-        this.eventBus.publish(new ZoneEvent(ctx, newZone, 'created', input));
+        await this.eventBus.publish(new ZoneEvent(ctx, newZone, 'created', input));
         return assertFound(this.findOne(ctx, newZone.id));
     }
 
@@ -129,7 +129,7 @@ export class ZoneService {
         const updatedZone = patchEntity(zone, input);
         await this.connection.getRepository(ctx, Zone).save(updatedZone, { reload: false });
         await this.zones.refresh(ctx);
-        this.eventBus.publish(new ZoneEvent(ctx, zone, 'updated', input));
+        await this.eventBus.publish(new ZoneEvent(ctx, zone, 'updated', input));
         return assertFound(this.findOne(ctx, zone.id));
     }
 
@@ -168,7 +168,7 @@ export class ZoneService {
         } else {
             await this.connection.getRepository(ctx, Zone).remove(zone);
             await this.zones.refresh(ctx);
-            this.eventBus.publish(new ZoneEvent(ctx, deletedZone, 'deleted', id));
+            await this.eventBus.publish(new ZoneEvent(ctx, deletedZone, 'deleted', id));
             return {
                 result: DeletionResult.DELETED,
                 message: '',
@@ -188,7 +188,7 @@ export class ZoneService {
         zone.members = members;
         await this.connection.getRepository(ctx, Zone).save(zone, { reload: false });
         await this.zones.refresh(ctx);
-        this.eventBus.publish(new ZoneMembersEvent(ctx, zone, 'assigned', memberIds));
+        await this.eventBus.publish(new ZoneMembersEvent(ctx, zone, 'assigned', memberIds));
         return assertFound(this.findOne(ctx, zone.id));
     }
 
@@ -202,7 +202,7 @@ export class ZoneService {
         zone.members = zone.members.filter(country => !memberIds.includes(country.id));
         await this.connection.getRepository(ctx, Zone).save(zone, { reload: false });
         await this.zones.refresh(ctx);
-        this.eventBus.publish(new ZoneMembersEvent(ctx, zone, 'removed', memberIds));
+        await this.eventBus.publish(new ZoneMembersEvent(ctx, zone, 'removed', memberIds));
         return assertFound(this.findOne(ctx, zone.id));
     }
 

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

@@ -197,7 +197,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
                     body: elasticSearchBody,
                 });
                 const totalItems = await this.totalHits(ctx, input, groupByProduct);
-                this.eventBus.publish(new SearchEvent(ctx, input));
+                await this.eventBus.publish(new SearchEvent(ctx, input));
                 return {
                     items: body.hits.hits.map(hit => this.mapProductToSearchResult(hit)),
                     totalItems,
@@ -222,7 +222,7 @@ export class ElasticsearchService implements OnModuleInit, OnModuleDestroy {
                     index: indexPrefix + VARIANT_INDEX_NAME,
                     body: elasticSearchBody,
                 });
-                this.eventBus.publish(new SearchEvent(ctx, input));
+                await this.eventBus.publish(new SearchEvent(ctx, input));
                 return {
                     items: body.hits.hits.map(hit => this.mapVariantToSearchResult(hit)),
                     totalItems: body.hits.total ? body.hits.total.value : 0,

packages/email-plugin/src/email-processor.ts

@@ -75,7 +75,7 @@ export class EmailProcessor {
             };
             const transportSettings = await this.getTransportSettings(ctx);
             await this.emailSender.send(emailDetails, transportSettings);
-            this.eventBus.publish(new EmailSendEvent(ctx, emailDetails, true));
+            await this.eventBus.publish(new EmailSendEvent(ctx, emailDetails, true));
             return true;
         } catch (err: unknown) {
             if (err instanceof Error) {
@@ -84,7 +84,7 @@ export class EmailProcessor {
                 Logger.error(String(err), loggerCtx);
             }
 
-            this.eventBus.publish(new EmailSendEvent(ctx, emailDetails, false, err as Error));
+            await this.eventBus.publish(new EmailSendEvent(ctx, emailDetails, false, err as Error));
             throw err;
         }
     }