fix(api): temporary timeout for EventsSubscription and isolate tests of register-course-user.test-module #368 (#369)

Co-authored-by: Hubert Kawałek <hubert.kawalek.dev@gmail.com>

Author: MateuszNaKodach

packages/api/src/module/automation/when-email-confirmation-was-approved-then-complete-user-registration/email-confirmation-was-approved-event-handler.service.ts

@@ -1,4 +1,4 @@
-import { Injectable, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
+import { Injectable, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
 import { CommandBus } from '@nestjs/cqrs';
 
 import { ApplicationEvent } from '@/module/application-command-events';
@@ -12,7 +12,7 @@ import { EventsSubscription } from '@/write/shared/application/events-subscripti
 import { EventsSubscriptionsRegistry } from '@/write/shared/application/events-subscription/events-subscriptions-registry';
 
 @Injectable()
-export class EmailConfirmationWasApprovedEventHandler implements OnModuleInit, OnModuleDestroy {
+export class EmailConfirmationWasApprovedEventHandler implements OnApplicationBootstrap, OnModuleDestroy {
   private eventsSubscription: EventsSubscription;
 
   constructor(
@@ -21,7 +21,7 @@ export class EmailConfirmationWasApprovedEventHandler implements OnModuleInit, O
     private readonly eventsSubscriptionsFactory: EventsSubscriptionsRegistry,
   ) {}
 
-  async onModuleInit() {
+  async onApplicationBootstrap() {
     this.eventsSubscription = this.eventsSubscriptionsFactory
       .subscription('WhenEmailConfirmationWasApprovedThenCompleteUserRegistrationAutomationModule_Automation_v1')
       .onEvent<EmailConfirmationWasApproved>('EmailConfirmationWasApproved', (event) =>

packages/api/src/module/automation/when-email-confirmation-was-approved-then-complete-user-registration/when-email-confirmation-was-approved-then-complete-user-registration.spec.ts

@@ -29,6 +29,6 @@ describe('CompleteUser registration when emailConfirmationWasApproved', () => {
       event,
     );
 
-    moduleUnderTest.expectCommandExecutedLastly(completeUserRegistrationCommand({ userId }));
+    await moduleUnderTest.expectCommandExecutedLastly(completeUserRegistrationCommand({ userId }));
   });
 });

packages/api/src/module/automation/when-user-registration-was-started-then-request-email-confirmation/user-registration-was-started.event-handler.service.ts

@@ -1,4 +1,4 @@
-import {Injectable, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
+import { Injectable, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
 import { CommandBus } from '@nestjs/cqrs';
 
 import { ApplicationEvent } from '@/module/application-command-events';

packages/api/src/module/read/course-progress/course-progress.read-module.ts

@@ -1,4 +1,4 @@
-import {Module, OnApplicationBootstrap, OnModuleDestroy} from '@nestjs/common';
+import { Module, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
 
 import { LearningMaterialsUrlWasGenerated } from '@/events/learning-materials-url-was-generated.domain-event';
 import { ApplicationEvent } from '@/module/application-command-events';

packages/api/src/module/read/learning-materials/learning-materials.read-module.ts

@@ -1,4 +1,4 @@
-import {Module, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
+import { Module, OnApplicationBootstrap, OnModuleDestroy } from '@nestjs/common';
 
 import { LearningMaterialsUrlWasGenerated } from '@/events/learning-materials-url-was-generated.domain-event';
 import { ApplicationEvent } from '@/module/application-command-events';

packages/api/src/module/write/register-course-user/register-course-user.test-module.ts

@@ -1,5 +1,8 @@
 import { initWriteTestModule } from '@/shared/test-utils';
+import { RegisterCourseUserWriteModule } from '@/write/register-course-user/register-course-user.write-module';
 
 export async function registerCourseUserTestModule() {
-  return initWriteTestModule();
+  return initWriteTestModule({
+    modules: [RegisterCourseUserWriteModule],
+  });
 }

packages/api/src/module/write/register-course-user/register-course-user.write-module.spec.ts

@@ -14,6 +14,14 @@ describe('registerCourseUserModule', () => {
     data: { userId, courseUserId, courseId },
   });
 
+  beforeEach(async () => {
+    module = await registerCourseUserTestModule();
+  });
+
+  afterEach(async () => {
+    await module.close();
+  });
+
   it('should register new course user', async () => {
     // Given
     const command = commandBuilder();
@@ -39,8 +47,4 @@ describe('registerCourseUserModule', () => {
     // Then
     await expect(() => module.executeCommand(() => command)).rejects.toThrow();
   });
-
-  beforeEach(async () => {
-    module = await registerCourseUserTestModule();
-  });
 });

packages/api/src/module/write/shared/application/events-subscription/events-subscription.ts

@@ -89,19 +89,26 @@ export class EventsSubscription {
    * See handleEvent for more details how handling event working.
    */
   async start(): Promise<void> {
-    const retryConfig = this.configuration.options?.retry ?? {
-      retries: 'INFINITELY',
-      backoff: 'EXPONENTIAL',
-      delay: 3000,
+    const retryConfig: Partial<RetryConfig<void>> = {
+      ...(this.configuration.options?.retry ?? {
+        retries: 'INFINITELY',
+        backoff: 'EXPONENTIAL',
+        delay: 3000,
+      }),
+      logger: (msg) => this.logger.error(msg),
+      // FIXME: after timeout EventsSubscription will stop processing
+      timeout: 2147483647, // max timeout => max 32-bit signed integer
     };
 
-    // TODO: extract failure recovery to separeate module which will run in bootstrap phase (SRP)
-    await retry(async () => {
+    this.eventEmitter.onAny(this.eventEmitterListener);
+
+    // FIXME: it will work for 2147483647/(1000*60*60*24) = 24.85 days :D
+    retry(async () => {
       await this.catchUp().catch((e) => {
         this.logger.warn(`EventsSubscription ${this.subscriptionId} processing error in CatchUp phase.`, e);
         throw e;
       });
-      this.listen().catch((e) => {
+      await this.listen().catch(async (e) => {
         this.logger.warn(`EventsSubscription ${this.subscriptionId} processing error in listen phase.`, e);
         throw e;
       });
@@ -118,7 +125,7 @@ export class EventsSubscription {
    */
   async stop(): Promise<void> {
     this.eventEmitter.offAny(this.eventEmitterListener);
-    this.queue.stop();
+    this.queue.clear();
   }
 
   /**
@@ -128,14 +135,11 @@ export class EventsSubscription {
    */
   async reset(): Promise<void> {
     await this.stop();
-    this.queue.clear();
     await this.moveCurrentPosition(this.configuration.options.start.from.globalPosition - 1, this.prismaService);
     await this.start();
   }
 
   private async listen(): Promise<void> {
-    this.eventEmitter.onAny(this.eventEmitterListener);
-
     let event = await this.queue.pop();
 
     while (!OrderedEventQueue.isStopToken(event)) {
@@ -172,10 +176,9 @@ export class EventsSubscription {
         : this.configuration.options.start.from.globalPosition,
     });
 
-    // eslint-disable-next-line no-restricted-syntax
-    for (const event of eventsToCatchup) {
-      await this.handleEvent(event);
-    }
+    this.eventsRetryCount.clear();
+    this.queue.clear();
+    eventsToCatchup.forEach((event) => this.queue.push(event));
   }
 
   /**
@@ -193,12 +196,10 @@ export class EventsSubscription {
     this.eventsRetryCount.delete(event.id);
 
     try {
-      // TODO add transaction
       await this.processSubscriptionPositionChange(event, this.prismaService);
       await this.processEvent(event, this.prismaService);
       await this.moveCurrentPosition(event.globalOrder, this.prismaService);
     } catch (e) {
-      await this.stop();
       this.logger.warn(
         `EventsSubscription ${this.subscriptionId} processing stopped on position ${event.globalOrder}`,
         e,