feat: delete queues when app exit

package.json

@@ -1,6 +1,6 @@
 {
     "name": "vtru-studio-websocket",
-    "version": "1.1.36",
+    "version": "1.1.37",
     "description": "",
     "main": "index.js",
     "scripts": {

src/controllers/notify/index.ts

@@ -1,7 +1,7 @@
 import debug from 'debug';
 import { nanoid } from 'nanoid';
 import { RABBITMQ_EXCHANGE_CREATORS } from '../../constants';
-import { getChannel } from '../../services/rabbitmq';
+import { disconnect, getChannel } from '../../services/rabbitmq';
 import { captureException } from '../../services/sentry';
 import { io } from '../../services';
 import { NotifyEnvelope } from './types';
@@ -28,19 +28,14 @@ export const start = async () => {
 
     logger('Channel controller notify started');
 
-    const notificationQueue = `${RABBITMQ_EXCHANGE_CREATORS}.notifications.${uniqueId}`;
-
+    const logQueue = `${RABBITMQ_EXCHANGE_CREATORS}.notifications.${uniqueId}`;
+    logger('logQueue', logQueue);
     channel.assertExchange(RABBITMQ_EXCHANGE_CREATORS, 'topic', {
         durable: true,
     });
-    channel.assertQueue(notificationQueue, { durable: false });
-    channel.bindQueue(
-        notificationQueue,
-        RABBITMQ_EXCHANGE_CREATORS,
-        'userNotification'
-    );
-
-    channel.consume(notificationQueue, async (message) => {
+    channel.assertQueue(logQueue, { durable: false });
+    channel.bindQueue(logQueue, RABBITMQ_EXCHANGE_CREATORS, 'userNotification');
+    channel.consume(logQueue, async (message) => {
         if (!message) return;
 
         try {
@@ -68,4 +63,12 @@ export const start = async () => {
         }
         channel.nack(message);
     });
+
+    process.once('SIGINT', async () => {
+        logger(`Deleting queue ${logQueue}`);
+        await channel.deleteQueue(logQueue);
+
+        // disconnect from RabbitMQ
+        await disconnect();
+    });
 };

src/controllers/preSignedURL/index.ts

@@ -2,7 +2,7 @@ import debug from 'debug';
 import { nanoid } from 'nanoid';
 import type { AssetEnvelope } from '../types';
 import { RABBITMQ_EXCHANGE_CREATORS } from '../../constants';
-import { getChannel } from '../../services/rabbitmq';
+import { disconnect, getChannel } from '../../services/rabbitmq';
 import { captureException } from '../../services/sentry';
 import { io } from '../../services';
 
@@ -30,15 +30,12 @@ export const start = async () => {
     logger('Channel controller preSignedURL started');
 
     const logQueue = `${RABBITMQ_EXCHANGE_CREATORS}.assets.${uniqueId}`;
-
+    logger('logQueue', logQueue);
     channel.assertExchange(RABBITMQ_EXCHANGE_CREATORS, 'topic', {
         durable: true,
     });
-
     channel.assertQueue(logQueue, { durable: false });
-
     channel.bindQueue(logQueue, RABBITMQ_EXCHANGE_CREATORS, 'preSignedURL');
-
     channel.consume(logQueue, async (message) => {
         console.log('message received:', message);
         if (!message) return;
@@ -78,4 +75,12 @@ export const start = async () => {
         }
         channel.nack(message);
     });
+
+    process.once('SIGINT', async () => {
+        logger(`Deleting queue ${logQueue}`);
+        await channel.deleteQueue(logQueue);
+
+        // disconnect from RabbitMQ
+        await disconnect();
+    });
 };

src/services/rabbitmq/index.ts

@@ -33,10 +33,16 @@ export const disconnect = async () => {
 
         try {
             await oldConnection.close();
+            return; // exit function
         } catch (error) {
-            // ignore
+            logger('Error closing RabbitMQ connection: %O', error);
+            captureException(error, { tags: { scope: 'rabbitmq' } });
+            process.exit(1);
         }
     }
+
+    logger('RabbitMQ connection not established');
+    process.exit(1);
 };
 
 export const getConnection = async () => {