refactor: consolidate queue services and modules

Author: kastov

libs/contract/constants/events/events.ts

@@ -9,12 +9,12 @@ export const EVENTS = {
         LIMITED: 'user.limited',
         EXPIRED: 'user.expired',
         TRAFFIC_RESET: 'user.traffic_reset',
-        EXPIRE_NOTIFY: {
-            EXPIRES_IN_72_HOURS: 'user.expires_in_72_hours',
-            EXPIRES_IN_48_HOURS: 'user.expires_in_48_hours',
-            EXPIRES_IN_24_HOURS: 'user.expires_in_24_hours',
-            EXPIRED_24_HOURS_AGO: 'user.expired_24_hours_ago',
-        },
+
+        EXPIRE_NOTIFY_EXPIRES_IN_72_HOURS: 'user.expires_in_72_hours',
+        EXPIRE_NOTIFY_EXPIRES_IN_48_HOURS: 'user.expires_in_48_hours',
+        EXPIRE_NOTIFY_EXPIRES_IN_24_HOURS: 'user.expires_in_24_hours',
+        EXPIRE_NOTIFY_EXPIRED_24_HOURS_AGO: 'user.expired_24_hours_ago',
+
         FIRST_CONNECTED: 'user.first_connected',
         BANDWIDTH_USAGE_THRESHOLD_REACHED: 'user.bandwidth_usage_threshold_reached',
         /**

libs/contract/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@remnawave/backend-contract",
-  "version": "2.3.28",
+  "version": "2.3.29",
   "public": true,
   "license": "AGPL-3.0-only",
   "description": "A contract library for Remnawave Backend. It can be used in backend and frontend.",

package-lock.json

@@ -88,7 +88,7 @@
     "@willsoto/nestjs-prometheus": "^6.0.2",
     "arctic": "^3.7.0",
     "axios": "^1.13.2",
-    "bullmq": "5.63.2",
+    "bullmq": "5.65.0",
     "cache-manager": "^7.2.5",
     "class-transformer": "^0.5.1",
     "compression": "^1.8.1",
@@ -141,7 +141,7 @@
     "zod": "^3.25.76"
   },
   "devDependencies": {
-    "@nestjs/cli": "11.0.13",
+    "@nestjs/cli": "11.0.14",
     "@nestjs/schematics": "11.0.9",
     "@types/compression": "^1.8.1",
     "@types/cookie-parser": "^1.4.10",

package.json

@@ -2,4 +2,5 @@ export * from './certs';
 export * from './convert-type';
 export * from './happ-crypto-link';
 export * from './mask-string';
+export * from './md5';
 export * from './superjson';

src/common/utils/index.ts

@@ -0,0 +1,5 @@
+import { createHash } from 'node:crypto';
+
+export function md5(data: string): string {
+    return createHash('md5').update(data).digest('hex');
+}

src/common/utils/md5.ts

@@ -179,7 +179,7 @@ export class UsersEvents {
         });
     }
 
-    @OnEvent(EVENTS.USER.EXPIRE_NOTIFY.EXPIRES_IN_72_HOURS)
+    @OnEvent(EVENTS.USER.EXPIRE_NOTIFY_EXPIRES_IN_72_HOURS)
     @RequireAdminId()
     async onUserExpiresIn72Hours(event: UserEvent): Promise<void> {
         const msg = `
@@ -194,7 +194,7 @@ export class UsersEvents {
         });
     }
 
-    @OnEvent(EVENTS.USER.EXPIRE_NOTIFY.EXPIRES_IN_48_HOURS)
+    @OnEvent(EVENTS.USER.EXPIRE_NOTIFY_EXPIRES_IN_48_HOURS)
     @RequireAdminId()
     async onUserExpiresIn48Hours(event: UserEvent): Promise<void> {
         const msg = `
@@ -209,7 +209,7 @@ export class UsersEvents {
         });
     }
 
-    @OnEvent(EVENTS.USER.EXPIRE_NOTIFY.EXPIRES_IN_24_HOURS)
+    @OnEvent(EVENTS.USER.EXPIRE_NOTIFY_EXPIRES_IN_24_HOURS)
     @RequireAdminId()
     async onUserExpiresIn24Hours(event: UserEvent): Promise<void> {
         const msg = `
@@ -224,7 +224,7 @@ export class UsersEvents {
         });
     }
 
-    @OnEvent(EVENTS.USER.EXPIRE_NOTIFY.EXPIRED_24_HOURS_AGO)
+    @OnEvent(EVENTS.USER.EXPIRE_NOTIFY_EXPIRED_24_HOURS_AGO)
     @RequireAdminId()
     async onUserExpired24HoursAgo(event: UserEvent): Promise<void> {
         const msg = `

src/integration-modules/notifications/telegram-bot/events/users/users.events.ts

@@ -9,8 +9,7 @@ import { ICommandResponse } from '@common/types/command-response.type';
 import { XRayConfig } from '@common/helpers/xray-config';
 import { ERRORS } from '@libs/contracts/constants/errors';
 
-import { StartAllNodesByProfileQueueService } from '@queue/start-all-nodes-by-profile';
-import { StopNodeQueueService } from '@queue/stop-node';
+import { NodesQueuesService } from '@queue/_nodes';
 
 import { GetConfigProfileByUuidResponseModel } from './models/get-config-profile-by-uuid.response.model';
 import { DeleteConfigProfileByUuidResponseModel, GetAllInboundsResponseModel } from './models';
@@ -28,8 +27,7 @@ export class ConfigProfileService {
 
     constructor(
         private readonly configProfileRepository: ConfigProfileRepository,
-        private readonly startAllNodesByProfileQueueService: StartAllNodesByProfileQueueService,
-        private readonly stopNodeQueueService: StopNodeQueueService,
+        private readonly nodesQueuesService: NodesQueuesService,
         private readonly queryBus: QueryBus,
     ) {}
 
@@ -145,7 +143,7 @@ export class ConfigProfileService {
             }
 
             for (const node of configProfile.nodes) {
-                await this.stopNodeQueueService.stopNode({
+                await this.nodesQueuesService.stopNode({
                     nodeUuid: node.uuid,
                     isNeedToBeDeleted: false,
                 });
@@ -264,7 +262,7 @@ export class ConfigProfileService {
                 // No need for now
                 // await this.commandBus.execute(new SyncActiveProfileCommand());
 
-                await this.startAllNodesByProfileQueueService.startAllNodesByProfile({
+                await this.nodesQueuesService.startAllNodesByProfile({
                     profileUuid: existingConfigProfile.uuid,
                     emitter: 'updateConfigProfile',
                 });

src/modules/config-profiles/config-profile.service.ts

@@ -10,7 +10,7 @@ import { ICommandResponse } from '@common/types/command-response.type';
 import { CACHE_KEYS, TSubscriptionTemplateType } from '@libs/contracts/constants';
 import { ERRORS } from '@libs/contracts/constants/errors';
 
-import { ExternalSquadActionsQueueService } from '@queue/external-squad-actions';
+import { SquadsQueueService } from '@queue/_squads';
 
 import {
     DeleteExternalSquadByUuidResponseModel,
@@ -28,7 +28,7 @@ export class ExternalSquadService {
 
     constructor(
         private readonly externalSquadRepository: ExternalSquadRepository,
-        private readonly externalSquadActionsQueueService: ExternalSquadActionsQueueService,
+        private readonly squadsQueueService: SquadsQueueService,
         @Inject(CACHE_MANAGER) private cacheManager: Cache,
     ) {}
 
@@ -251,7 +251,7 @@ export class ExternalSquadService {
                 };
             }
 
-            await this.externalSquadActionsQueueService.addUsersToExternalSquad({
+            await this.squadsQueueService.addUsersToExternalSquad({
                 externalSquadUuid: uuid,
             });
 
@@ -281,7 +281,7 @@ export class ExternalSquadService {
                 };
             }
 
-            await this.externalSquadActionsQueueService.removeUsersFromExternalSquad({
+            await this.squadsQueueService.removeUsersFromExternalSquad({
                 externalSquadUuid: uuid,
             });
 

src/modules/external-squads/external-squads.service.ts

@@ -6,8 +6,8 @@ import { Injectable, Logger } from '@nestjs/common';
 import { ICommandResponse } from '@common/types/command-response.type';
 import { ERRORS } from '@libs/contracts/constants/errors';
 
-import { StartAllNodesByProfileQueueService } from '@queue/start-all-nodes-by-profile';
-import { InternalSquadActionsQueueService } from '@queue/internal-squad-actions';
+import { SquadsQueueService } from '@queue/_squads';
+import { NodesQueuesService } from '@queue/_nodes';
 
 import { GetInternalSquadByUuidResponseModel } from './models/get-internal-squad-by-uuid.response.model';
 import { DeleteInternalSquadResponseModel } from './models/delete-internal-squad-by-uuid.response.model';
@@ -24,8 +24,8 @@ export class InternalSquadService {
 
     constructor(
         private readonly internalSquadRepository: InternalSquadRepository,
-        private readonly startAllNodesByProfileQueueService: StartAllNodesByProfileQueueService,
-        private readonly internalSquadActionsQueueService: InternalSquadActionsQueueService,
+        private readonly nodesQueuesService: NodesQueuesService,
+        private readonly squadsQueueService: SquadsQueueService,
     ) {}
 
     public async getInternalSquads(): Promise<ICommandResponse<GetInternalSquadsResponseModel>> {
@@ -197,7 +197,7 @@ export class InternalSquadService {
 
                     await Promise.all(
                         affectedConfigProfiles.map((profileUuid) =>
-                            this.startAllNodesByProfileQueueService.startAllNodesByProfile({
+                            this.nodesQueuesService.startAllNodesByProfile({
                                 profileUuid,
                                 emitter: 'updateInternalSquad',
                             }),
@@ -264,7 +264,7 @@ export class InternalSquadService {
             const deleted = await this.internalSquadRepository.deleteByUUID(uuid);
 
             for (const profileUuid of includedProfiles) {
-                await this.startAllNodesByProfileQueueService.startAllNodesByProfile({
+                await this.nodesQueuesService.startAllNodesByProfile({
                     profileUuid,
                     emitter: 'deleteInternalSquad',
                 });
@@ -296,7 +296,7 @@ export class InternalSquadService {
                 };
             }
 
-            await this.internalSquadActionsQueueService.addUsersToInternalSquad({
+            await this.squadsQueueService.addUsersToInternalSquad({
                 internalSquadUuid: uuid,
             });
 
@@ -326,7 +326,7 @@ export class InternalSquadService {
                 };
             }
 
-            await this.internalSquadActionsQueueService.removeUsersFromInternalSquad({
+            await this.squadsQueueService.removeUsersFromInternalSquad({
                 internalSquadUuid: uuid,
             });