feat: add node health check queue and refactor health check service

- Introduce NodeHealthCheckQueueModule and NodeHealthCheckQueueService
- Simplify NodeHealthCheckService by delegating node health checks to queue
- Remove unused event and command bus dependencies
- Add new queue name for node health check in QueueNames enum
- Update queue module to include NodeHealthCheckQueueModule

Author: kastov

.hygen/new/queue/queue.module.ejs.t

@@ -16,6 +16,7 @@ unless_exists: true
 import { BullBoardModule } from '@bull-board/nestjs';
 
 import { BullModule } from '@nestjs/bullmq';
+import { CqrsModule } from '@nestjs/cqrs';
 import { Module } from '@nestjs/common';
 
 import { useBullBoard, useQueueProcessor } from '@common/utils/startup-app';

.hygen/new/queue/queue.processor.ejs.t

@@ -11,6 +11,7 @@ unless_exists: true
 %>import { Job } from 'bullmq';
 
 import { Processor, WorkerHost } from '@nestjs/bullmq';
+import { CommandBus, QueryBus } from '@nestjs/cqrs';
 import { Logger } from '@nestjs/common';
 
 import { <%= QueueJobNamesEnumName %> } from './enums';
@@ -20,12 +21,19 @@ import { QueueNames } from '../queue.enum';
 export class <%= QueueProcessorName %> extends WorkerHost {
     private readonly logger = new Logger(<%= QueueProcessorName %>.name)
 
+    constructor(
+        private readonly queryBus: QueryBus,
+        private readonly commandBus: CommandBus,
+    ) {
+        super();
+    }
+
     async process(job: Job) {
         switch (job.name) {
             case <%= QueueJobNamesEnumName %>.exampleJob:
                 return this.handleExampleJob(job);
             default:
-                this.logger.warn(`⚠️ Job "${job.name}" is not handled.`);
+                this.logger.warn(`🚨 Job "${job.name}" is not handled.`);
                 break;
         }
     }

src/modules/xray-config/xray-config.service.ts

@@ -1,7 +1,7 @@
 import { ERRORS } from '@contract/constants';
 
 import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common';
-import { CommandBus, EventBus, QueryBus } from '@nestjs/cqrs';
+import { CommandBus, QueryBus } from '@nestjs/cqrs';
 
 import { ICommandResponse } from '@common/types/command-response.type';
 import { IXrayConfig } from '@common/helpers/xray-config/interfaces';
@@ -27,7 +27,6 @@ export class XrayConfigService implements OnApplicationBootstrap {
     constructor(
         private readonly commandBus: CommandBus,
         private readonly queryBus: QueryBus,
-        private readonly eventBus: EventBus,
         private readonly xrayConfigRepository: XrayConfigRepository,
         private readonly startAllNodesQueue: StartAllNodesQueueService,
     ) {}

src/queue/node-health-check/enums/index.ts

@@ -0,0 +1 @@
+export * from './node-health-check-job-names.enum';

src/queue/node-health-check/enums/node-health-check-job-names.enum.ts

@@ -0,0 +1,3 @@
+export enum NodeHealthCheckJobNames {
+    checkNodeHealth = 'checkNodeHealth',
+}

src/queue/node-health-check/index.ts

@@ -0,0 +1,2 @@
+export * from './enums';
+export * from './node-health-check.service';

src/queue/node-health-check/interfaces/index.ts

@@ -0,0 +1 @@
+export * from './node-health-check.interface';

src/queue/node-health-check/interfaces/node-health-check.interface.ts

@@ -0,0 +1,5 @@
+export interface NodeHealthCheckPayload {
+    nodeUuid: string;
+    nodeAddress: string;
+    nodePort: number | null;
+}

src/queue/node-health-check/node-health-check.module.ts

@@ -0,0 +1,33 @@
+import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
+import { BullBoardModule } from '@bull-board/nestjs';
+
+import { BullModule } from '@nestjs/bullmq';
+import { CqrsModule } from '@nestjs/cqrs';
+import { Module } from '@nestjs/common';
+
+import { useBullBoard, useQueueProcessor } from '@common/utils/startup-app';
+
+import { NodeHealthCheckQueueProcessor } from './node-health-check.processor';
+import { NodeHealthCheckQueueService } from './node-health-check.service';
+import { QueueNames } from '../queue.enum';
+
+const requiredModules = [CqrsModule];
+
+const processors = [NodeHealthCheckQueueProcessor];
+const services = [NodeHealthCheckQueueService];
+
+const queues = [BullModule.registerQueue({ name: QueueNames.nodeHealthCheck })];
+
+const bullBoard = [
+    BullBoardModule.forFeature({ name: QueueNames.nodeHealthCheck, adapter: BullMQAdapter }),
+];
+
+const providers = useQueueProcessor() ? processors : [];
+const imports = useBullBoard() ? bullBoard : [];
+
+@Module({
+    imports: [...queues, ...imports, ...requiredModules],
+    providers: [...providers, ...services],
+    exports: [...services],
+})
+export class NodeHealthCheckQueueModule {}

src/queue/node-health-check/node-health-check.processor.ts

@@ -0,0 +1,129 @@
+import { Job } from 'bullmq';
+
+import { Processor, WorkerHost } from '@nestjs/bullmq';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { CommandBus, QueryBus } from '@nestjs/cqrs';
+import { Logger } from '@nestjs/common';
+
+import { GetSystemStatsCommand } from '@remnawave/node-contract';
+
+import { ICommandResponse } from '@common/types/command-response.type';
+import { AxiosService } from '@common/axios';
+import { EVENTS } from '@libs/contracts/constants';
+
+import { NodeEvent } from '@intergration-modules/telegram-bot/events/nodes/interfaces';
+
+import { GetEnabledNodesQuery } from '@modules/nodes/queries/get-enabled-nodes';
+import { UpdateNodeCommand } from '@modules/nodes/commands/update-node';
+import { NodesEntity } from '@modules/nodes/entities/nodes.entity';
+
+import { NodeHealthCheckJobNames } from './enums';
+import { QueueNames } from '../queue.enum';
+
+@Processor(QueueNames.nodeHealthCheck, {
+    concurrency: 40,
+})
+export class NodeHealthCheckQueueProcessor extends WorkerHost {
+    private readonly logger = new Logger(NodeHealthCheckQueueProcessor.name);
+
+    constructor(
+        private readonly queryBus: QueryBus,
+        private readonly commandBus: CommandBus,
+        private readonly eventEmitter: EventEmitter2,
+        private readonly axios: AxiosService,
+    ) {
+        super();
+    }
+    async process(job: Job) {
+        switch (job.name) {
+            case NodeHealthCheckJobNames.checkNodeHealth:
+                return this.handleCheckNodeHealthJob(job);
+            default:
+                this.logger.warn(`🚨 Job "${job.name}" is not handled.`);
+                break;
+        }
+    }
+
+    private async handleCheckNodeHealthJob(job: Job<NodesEntity>) {
+        this.logger.debug(
+            `✅ Handling "${NodeHealthCheckJobNames.checkNodeHealth}" job with ID: ${job?.id || ''}, data: ${JSON.stringify(job?.data || '')}`,
+        );
+
+        try {
+            const response = await this.axios.getSystemStats(job.data.address, job.data.port);
+            switch (response.isOk) {
+                case true:
+                    return this.handleConnectedNode(job.data, response.response!);
+                case false:
+                    return this.handleDisconnectedNode(job.data, response.message);
+            }
+        } catch (error) {
+            this.logger.error(
+                `❌ Error handling "${NodeHealthCheckJobNames.checkNodeHealth}" job: ${error}`,
+            );
+        }
+    }
+
+    private async handleConnectedNode(node: NodesEntity, response: GetSystemStatsCommand.Response) {
+        if (typeof response.response.uptime !== 'number') {
+            this.logger.error(`Node ${node.uuid} uptime is not a number`);
+            return;
+        }
+
+        await this.updateNode({
+            node: {
+                uuid: node.uuid,
+                isConnected: true,
+                isNodeOnline: true,
+                isXrayRunning: true,
+                lastStatusChange: new Date(),
+                lastStatusMessage: '',
+            },
+        });
+
+        if (!node.isConnected) {
+            this.eventEmitter.emit(
+                EVENTS.NODE.CONNECTION_RESTORED,
+                new NodeEvent(node, EVENTS.NODE.CONNECTION_RESTORED),
+            );
+        }
+    }
+
+    private async handleDisconnectedNode(node: NodesEntity, message: string | undefined) {
+        this.logger.debug(`Node ${node.uuid} is disconnected: ${message}`);
+
+        const newNodeEntity = await this.updateNode({
+            node: {
+                uuid: node.uuid,
+                isConnected: false,
+                isNodeOnline: false,
+                isXrayRunning: false,
+                lastStatusChange: new Date(),
+                lastStatusMessage: message,
+                usersOnline: 0,
+            },
+        });
+
+        // this.eventBus.publish(new StartNodeEvent(newNodeEntity.response || node));
+
+        if (node.isConnected) {
+            node.lastStatusMessage = message || null;
+            this.eventEmitter.emit(
+                EVENTS.NODE.CONNECTION_LOST,
+                new NodeEvent(node, EVENTS.NODE.CONNECTION_LOST),
+            );
+        }
+    }
+
+    private async getEnabledNodes(): Promise<ICommandResponse<NodesEntity[]>> {
+        return this.queryBus.execute<GetEnabledNodesQuery, ICommandResponse<NodesEntity[]>>(
+            new GetEnabledNodesQuery(),
+        );
+    }
+
+    private async updateNode(dto: UpdateNodeCommand): Promise<ICommandResponse<NodesEntity>> {
+        return this.commandBus.execute<UpdateNodeCommand, ICommandResponse<NodesEntity>>(
+            new UpdateNodeCommand(dto.node),
+        );
+    }
+}