Merge pull request #142 from blockful/dev

Dev

Author: LeonardoVieira1630

apps/consumers/src/app.ts

@@ -47,6 +47,7 @@ export class App {
     if (this.rabbitmqConsumerService) {
       await this.rabbitmqConsumerService.stop();
     }
+    
     this.telegramBotService.stop('SIGINT');
   }
 } 

apps/consumers/src/services/rabbitmq-notification-consumer.service.ts

@@ -35,14 +35,20 @@ export class RabbitMQNotificationConsumerService {
   }
 
   async stop(): Promise<void> {
-    await this.consumer.close();
-    await this.connection.close();
+    if (this.consumer) {
+      await this.consumer.close();
+    }
+    
+    if (this.connection) {
+      await this.connection.close();
+    }
   }
 
   private async processNotification(message: RabbitMQMessage<NotificationPayload>): Promise<void> {
     if (message.type !== 'NOTIFICATION_EVENT') {
       return;
     }
+    
     try {
       await this.telegramBotService.sendNotification(message.payload);
     } catch (error: any) {

apps/dispatcher/src/app.ts

@@ -14,6 +14,7 @@ import { AnticaptureClient } from '@notification-system/anticapture-client';
 export class App {
   private rabbitMQConsumerService!: RabbitMQConsumerService;
   private rabbitmqConnection!: RabbitMQConnection;
+  private publisher!: RabbitMQPublisher;
   private isCreated = false;
 
   constructor(
@@ -45,9 +46,9 @@ export class App {
 
     this.rabbitmqConnection = new RabbitMQConnection(this.rabbitmqUrl);
     await this.rabbitmqConnection.connect();
-    const publisher = await RabbitMQPublisher.create(this.rabbitmqConnection);
+    this.publisher = await RabbitMQPublisher.create(this.rabbitmqConnection);
     const notificationFactory = new NotificationClientFactory();
-    notificationFactory.addClient('telegram', new RabbitMQNotificationService(publisher));
+    notificationFactory.addClient('telegram', new RabbitMQNotificationService(this.publisher));
     const triggerProcessorService = new TriggerProcessorService();
 
     triggerProcessorService.addHandler(
@@ -82,7 +83,16 @@ export class App {
   }
 
   async stop(): Promise<void> {
-    await this.rabbitMQConsumerService?.stop();
-    await this.rabbitmqConnection?.close();
+    if (this.rabbitMQConsumerService) {
+      await this.rabbitMQConsumerService.stop();
+    }
+    
+    if (this.publisher) {
+      await this.publisher.close();
+    }
+    
+    if (this.rabbitmqConnection) {
+      await this.rabbitmqConnection.close();
+    }
   }
 } 

apps/dispatcher/src/services/rabbitmq-consumer.service.ts

@@ -23,7 +23,12 @@ export class RabbitMQConsumerService {
   }
 
   async stop(): Promise<void> {
-    await this.consumer.close();
-    await this.connection.close();
+    if (this.consumer) {
+      await this.consumer.close();
+    }
+    
+    if (this.connection) {
+      await this.connection.close();
+    }
   }
 }

apps/dispatcher/src/services/triggers/non-voting-handler.service.test.ts

@@ -107,8 +107,9 @@ describe('NonVotingHandler', () => {
     expect(result.messageId).toBe('proposal-finished');
     expect(mockAnticaptureClient.listProposals).toHaveBeenCalledWith(
       expect.objectContaining({
-        status: expect.arrayContaining(['SUCCEEDED']),
-        limit: 3
+        status: expect.arrayContaining(['EXECUTED', 'SUCCEEDED', 'DEFEATED', 'EXPIRED', 'CANCELED']),
+        limit: 15,
+        orderDirection: 'desc'
       }),
       'ENS'
     );

apps/integrated-tests/jest.config.ts

@@ -8,6 +8,6 @@ export default {
   globalTeardown: '<rootDir>/src/setup/jest/jest-global-teardown.ts',
   setupFilesAfterEnv: ['<rootDir>/src/setup/jest/jest-setup-after-env.ts'],
   forceExit: true,
-  silent: true,
+  silent: false,
   verbose: true,
 }; 

apps/integrated-tests/package.json

@@ -15,14 +15,13 @@
     "@notification-system/rabbitmq-client": "workspace:*",
     "@notification-system/subscription-server": "workspace:*",
     "@testcontainers/rabbitmq": "^10.15.0",
-    "@types/amqplib": "^0.10.5",
     "@types/jest": "^29.5.14",
     "@types/node": "^20.17.46",
     "@types/uuid": "^9.0.0",
-    "amqplib": "^0.10.4",
     "dotenv": "^16.5.0",
     "jest": "^29.7.0",
     "knex": "^3.1.0",
+    "rabbitmq-client": "^5.0.5",
     "sqlite3": "^5.1.7",
     "ts-jest": "^29.3.2",
     "typescript": "^5.8.3",

apps/integrated-tests/src/config/services.ts

@@ -2,15 +2,18 @@
  * Service configuration for integration tests
  */
 
+// Generate random port to avoid conflicts in parallel tests
+const randomPort = 14000 + Math.floor(Math.random() * 1000);
+
 export const serviceConfig = {
   // Service ports 
   ports: {
-    subscriptionServer: 14001,
+    subscriptionServer: randomPort,
   },
 
   // Service URLs 
   urls: {
-    subscriptionServer: 'http://localhost:14001',
+    subscriptionServer: `http://localhost:${randomPort}`,
   },
 
   // Logic system configuration 

apps/integrated-tests/src/setup/jest/jest-global-setup.ts

@@ -13,10 +13,18 @@ declare global {
  */
 export default async function globalSetup() {
   const container = await new RabbitMQContainer()
-    .withStartupTimeout(150000)
+    .withStartupTimeout(30000)
     .start();
 
-  const amqpUrl = container.getAmqpUrl();
+  let amqpUrl = container.getAmqpUrl();
+  
+  // Fix URL to include credentials (testcontainers doesn't include them)
+  const urlObj = new URL(amqpUrl);
+  if (!urlObj.username && !urlObj.password) {
+    urlObj.username = 'guest';
+    urlObj.password = 'guest';
+  }
+  amqpUrl = urlObj.toString();
 
   // Store RabbitMQ URL in environment variable for tests to access
   process.env.TEST_RABBITMQ_URL = amqpUrl;

apps/integrated-tests/src/setup/rabbitmq-setup.ts

@@ -12,10 +12,11 @@ import { RabbitMQSetupConfig } from './types/rabbitmq-setup.types';
 class GlobalRabbitMQSetup {
   private containerManager = new RabbitMQContainerManager();
   private connectionManager = new RabbitMQConnectionManager();
-  private queueManager = new RabbitMQQueueManager(this.connectionManager);
-  private spyConsumerManager = new RabbitMQSpyConsumerManager(this.connectionManager);
+  private queueManager: RabbitMQQueueManager | null = null;
+  private spyConsumerManager: RabbitMQSpyConsumerManager | null = null;
   private eventCollector = new EventCollector();
   private isStarted = false;
+  private amqpUrl: string | null = null;
 
   async getOrCreateSetup(): Promise<RabbitMQSetupConfig> {
     if (this.isStarted) {
@@ -33,6 +34,12 @@ class GlobalRabbitMQSetup {
     // Setup container and connection
     const container = await this.containerManager.getContainer();
     const amqpUrl = container.getAmqpUrl();
+    this.amqpUrl = amqpUrl;
+    
+    // Initialize managers with the correct URL
+    this.queueManager = new RabbitMQQueueManager(this.connectionManager, amqpUrl);
+    this.spyConsumerManager = new RabbitMQSpyConsumerManager(this.connectionManager, amqpUrl);
+    
     await this.connectionManager.initialize(amqpUrl);
 
     // Setup initial queues and spy consumers
@@ -50,22 +57,30 @@ class GlobalRabbitMQSetup {
   }
 
   private async setupInitialQueues(): Promise<void> {
-    await this.queueManager.clearQueue('dispatcher-queue');
-    await this.spyConsumerManager.setupSpyConsumer({
-      queueName: 'dispatcher-queue',
-      eventCollector: this.eventCollector
-    });
+    if (this.queueManager && this.spyConsumerManager) {
+      await this.queueManager.clearQueue('dispatcher-queue');
+      await this.spyConsumerManager.setupSpyConsumer({
+        queueName: 'dispatcher-queue',
+        eventCollector: this.eventCollector
+      });
+    }
   }
 
   async globalCleanup(): Promise<void> {
     if (!this.isStarted) return;
 
-    await this.spyConsumerManager.cleanup();
+    if (this.spyConsumerManager) {
+      await this.spyConsumerManager.cleanup();
+    }
+    if (this.queueManager) {
+      await this.queueManager.cleanup();
+    }
     await this.connectionManager.cleanup();
     await this.containerManager.cleanup();
 
     this.isStarted = false;
     this.eventCollector.clear();
+    this.amqpUrl = null;
   }
 }
 
@@ -101,8 +116,8 @@ export class RabbitMQTestSetup {
     this.connectionManager = new RabbitMQConnectionManager();
     await this.connectionManager.initialize(amqpUrl);
 
-    this.queueManager = new RabbitMQQueueManager(this.connectionManager);
-    this.spyConsumerManager = new RabbitMQSpyConsumerManager(this.connectionManager);
+    this.queueManager = new RabbitMQQueueManager(this.connectionManager, amqpUrl);
+    this.spyConsumerManager = new RabbitMQSpyConsumerManager(this.connectionManager, amqpUrl);
 
     // Setup spy consumer for dispatcher queue
     await this.spyConsumerManager.setupSpyConsumer({
@@ -118,6 +133,7 @@ export class RabbitMQTestSetup {
 
     if (this.queueManager) {
       await this.queueManager.clearQueue('dispatcher-queue');
+      await this.queueManager.cleanup();
     }
 
     if (this.spyConsumerManager) {