refactor: node queries

kastov · kastov · commit 289d2eecd963 · 2025-12-04T18:44:31.000+03:00
diff --git a/src/modules/nodes/queries/get-enabled-nodes-partial/get-enabled-nodes-partial.handler.ts b/src/modules/nodes/queries/get-enabled-nodes-partial/get-enabled-nodes-partial.handler.ts
@@ -0,0 +1,28 @@
+import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
+import { Logger } from '@nestjs/common';
+
+import { fail, ok, TResult } from '@common/types';
+import { ERRORS } from '@libs/contracts/constants';
+
+import {
+    GetEnabledNodesPartialQuery,
+    IGetEnabledNodesPartialResponse,
+} from './get-enabled-nodes-partial.query';
+import { NodesRepository } from '../../repositories/nodes.repository';
+
+@QueryHandler(GetEnabledNodesPartialQuery)
+export class GetEnabledNodesPartialHandler implements IQueryHandler<GetEnabledNodesPartialQuery> {
+    private readonly logger = new Logger(GetEnabledNodesPartialHandler.name);
+    constructor(private readonly nodesRepository: NodesRepository) {}
+
+    async execute(): Promise<TResult<IGetEnabledNodesPartialResponse[]>> {
+        try {
+            const nodes = await this.nodesRepository.findEnabledNodesPartial();
+
+            return ok(nodes);
+        } catch (error) {
+            this.logger.error(error);
+            return fail(ERRORS.INTERNAL_SERVER_ERROR);
+        }
+    }
+}
diff --git a/src/modules/nodes/queries/get-enabled-nodes-partial/get-enabled-nodes-partial.query.ts b/src/modules/nodes/queries/get-enabled-nodes-partial/get-enabled-nodes-partial.query.ts
@@ -0,0 +1,16 @@
+import { Query } from '@nestjs/cqrs';
+
+import { TResult } from '@common/types';
+
+export interface IGetEnabledNodesPartialResponse {
+    uuid: string;
+    address: string;
+    port: number | null;
+    isConnected: boolean;
+}
+
+export class GetEnabledNodesPartialQuery extends Query<TResult<IGetEnabledNodesPartialResponse[]>> {
+    constructor() {
+        super();
+    }
+}
diff --git a/src/modules/nodes/queries/get-enabled-nodes-partial/index.ts b/src/modules/nodes/queries/get-enabled-nodes-partial/index.ts
@@ -0,0 +1,2 @@
+export * from './get-enabled-nodes-partial.handler';
+export * from './get-enabled-nodes-partial.query';
diff --git a/src/modules/nodes/queries/get-enabled-nodes/get-enabled-nodes.handler.ts b/src/modules/nodes/queries/get-enabled-nodes/get-enabled-nodes.handler.ts
@@ -1,22 +1,18 @@
 import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
 import { Logger } from '@nestjs/common';
 
-import { fail, ok, TResult } from '@common/types';
+import { fail, ok } from '@common/types';
 import { ERRORS } from '@libs/contracts/constants';
 
 import { NodesRepository } from '../../repositories/nodes.repository';
 import { GetEnabledNodesQuery } from './get-enabled-nodes.query';
-import { NodesEntity } from '../../entities/nodes.entity';
 
 @QueryHandler(GetEnabledNodesQuery)
-export class GetEnabledNodesHandler implements IQueryHandler<
-    GetEnabledNodesQuery,
-    TResult<NodesEntity[]>
-> {
+export class GetEnabledNodesHandler implements IQueryHandler<GetEnabledNodesQuery> {
     private readonly logger = new Logger(GetEnabledNodesHandler.name);
     constructor(private readonly nodesRepository: NodesRepository) {}
 
-    async execute(): Promise<TResult<NodesEntity[]>> {
+    async execute() {
         try {
             const nodes = await this.nodesRepository.findByCriteria({
                 isDisabled: false,
diff --git a/src/modules/nodes/queries/get-enabled-nodes/get-enabled-nodes.query.ts b/src/modules/nodes/queries/get-enabled-nodes/get-enabled-nodes.query.ts
@@ -1,3 +1,11 @@
-export class GetEnabledNodesQuery {
-    constructor() {}
+import { Query } from '@nestjs/cqrs';
+
+import { TResult } from '@common/types';
+
+import { NodesEntity } from '../../entities/nodes.entity';
+
+export class GetEnabledNodesQuery extends Query<TResult<NodesEntity[]>> {
+    constructor() {
+        super();
+    }
 }
diff --git a/src/modules/nodes/queries/index.ts b/src/modules/nodes/queries/index.ts
@@ -1,3 +1,4 @@
+import { GetEnabledNodesPartialHandler } from './get-enabled-nodes-partial/get-enabled-nodes-partial.handler';
 import { FindNodesByCriteriaHandler } from './find-nodes-by-criteria';
 import { GetNodesByCriteriaHandler } from './get-nodes-by-criteria';
 import { CountOnlineUsersHandler } from './count-online-users';
@@ -14,4 +15,5 @@ export const QUERIES = [
     CountOnlineUsersHandler,
     GetNodeByUuidHandler,
     FindNodesByCriteriaHandler,
+    GetEnabledNodesPartialHandler,
 ];
diff --git a/src/modules/nodes/repositories/nodes.repository.ts b/src/modules/nodes/repositories/nodes.repository.ts
@@ -9,6 +9,7 @@ import { getKyselyUuid } from '@common/helpers/kysely/get-kysely-uuid';
 import { TxKyselyService } from '@common/database';
 import { ICrud } from '@common/types/crud-port';
 
+import { IGetEnabledNodesPartialResponse } from '../queries/get-enabled-nodes-partial/get-enabled-nodes-partial.query';
 import { IGetOnlineNodesPartialResponse } from '../queries/get-online-nodes';
 import { NodesEntity } from '../entities/nodes.entity';
 import { NodesConverter } from '../nodes.converter';
@@ -81,6 +82,17 @@ export class NodesRepository implements ICrud<NodesEntity> {
         return nodesList;
     }
 
+    public async findEnabledNodesPartial(): Promise<IGetEnabledNodesPartialResponse[]> {
+        const nodesList = await this.qb.kysely
+            .selectFrom('nodes')
+            .select(['uuid', 'address', 'port', 'isConnected'])
+            .where('isDisabled', '=', false)
+            .where('isConnecting', '=', false)
+            .execute();
+
+        return nodesList;
+    }
+
     public async findConnectedNodesWithoutInbounds(): Promise<
         {
             uuid: string;
diff --git a/src/queue/_nodes/interfaces/node-health-check.interface.ts b/src/queue/_nodes/interfaces/node-health-check.interface.ts
@@ -3,5 +3,4 @@ export interface INodeHealthCheckPayload {
     nodeAddress: string;
     nodePort: number | null;
     isConnected: boolean;
-    isConnecting: boolean;
 }
diff --git a/src/queue/_nodes/nodes-queues.service.ts b/src/queue/_nodes/nodes-queues.service.ts
@@ -3,7 +3,7 @@ import { Queue } from 'bullmq';
 import { Injectable, Logger, OnApplicationBootstrap } from '@nestjs/common';
 import { InjectQueue } from '@nestjs/bullmq';
 
-import { NodesEntity } from '@modules/nodes';
+import { IGetEnabledNodesPartialResponse } from '@modules/nodes/queries/get-enabled-nodes-partial/get-enabled-nodes-partial.query';
 
 import { QUEUES_NAMES } from '@queue/queue.enum';
 
@@ -77,7 +77,7 @@ export class NodesQueuesService implements OnApplicationBootstrap {
         });
     }
 
-    public async checkNodeHealthBulk(payload: NodesEntity[]) {
+    public async checkNodeHealthBulk(payload: IGetEnabledNodesPartialResponse[]) {
         return this.nodeHealthCheckQueue.addBulk(
             payload.map((node) => {
                 return {
@@ -87,7 +87,6 @@ export class NodesQueuesService implements OnApplicationBootstrap {
                         nodeAddress: node.address,
                         nodePort: node.port,
                         isConnected: node.isConnected,
-                        isConnecting: node.isConnecting,
                     } satisfies INodeHealthCheckPayload,
                     opts: {
                         jobId: `${NODES_JOB_NAMES.NODE_HEALTH_CHECK}-${node.uuid}`,
diff --git a/src/queue/_nodes/processors/node-health-check.processor.ts b/src/queue/_nodes/processors/node-health-check.processor.ts
@@ -36,14 +36,7 @@ export class NodeHealthCheckQueueProcessor extends WorkerHost {
     }
     async process(job: Job<INodeHealthCheckPayload>) {
         try {
-            const { nodeAddress, nodePort, nodeUuid, isConnected, isConnecting } = job.data;
-
-            if (isConnecting) {
-                this.logger.warn(
-                    `Node ${nodeUuid}, ${nodeAddress}:${nodePort} – is connecting, skipping health check`,
-                );
-                return;
-            }
+            const { nodeAddress, nodePort, nodeUuid, isConnected } = job.data;
 
             const attemptsLimit = 2;
             let attempts = 0;
diff --git a/src/scheduler/enqueue/node-health-check/node-health-check.task.ts b/src/scheduler/enqueue/node-health-check/node-health-check.task.ts
@@ -2,10 +2,7 @@ import { Injectable, Logger } from '@nestjs/common';
 import { QueryBus } from '@nestjs/cqrs';
 import { Cron } from '@nestjs/schedule';
 
-import { TResult } from '@common/types';
-
-import { GetEnabledNodesQuery } from '@modules/nodes/queries/get-enabled-nodes';
-import { NodesEntity } from '@modules/nodes';
+import { GetEnabledNodesPartialQuery } from '@modules/nodes/queries/get-enabled-nodes-partial/get-enabled-nodes-partial.query';
 
 import { NodesQueuesService } from '@queue/_nodes';
 
@@ -43,22 +40,20 @@ export class NodeHealthCheckTask {
                 return;
             }
 
-            const nodesResponse = await this.getEnabledNodes();
+            const nodesResponse = await this.queryBus.execute(new GetEnabledNodesPartialQuery());
             if (!nodesResponse.isOk) {
                 return;
             }
 
+            if (nodesResponse.response.length === 0) {
+                return;
+            }
+
             await this.nodesQueuesService.checkNodeHealthBulk(nodesResponse.response);
 
             return;
         } catch (error) {
             this.logger.error(`Error in NodeHealthCheckService: ${error}`);
         }
     }
-
-    private async getEnabledNodes(): Promise<TResult<NodesEntity[]>> {
-        return this.queryBus.execute<GetEnabledNodesQuery, TResult<NodesEntity[]>>(
-            new GetEnabledNodesQuery(),
-        );
-    }
 }
diff --git a/src/scheduler/tasks/review-nodes/review-nodes.task.ts b/src/scheduler/tasks/review-nodes/review-nodes.task.ts
@@ -3,7 +3,6 @@ import { Injectable, Logger } from '@nestjs/common';
 import { Cron } from '@nestjs/schedule';
 import { QueryBus } from '@nestjs/cqrs';
 
-import { TResult } from '@common/types';
 import { EVENTS } from '@libs/contracts/constants/events/events';
 
 import { NodeEvent } from '@integration-modules/notifications/interfaces';
@@ -33,11 +32,16 @@ export class ReviewNodesTask {
     async handleCron() {
         let nodes: NodesEntity[] | null = null;
         try {
-            const nodesResponse = await this.getEnabledNodes();
+            const nodesResponse = await this.queryBus.execute(new GetEnabledNodesQuery());
+
             if (!nodesResponse.isOk) {
                 return;
             }
 
+            if (nodesResponse.response.length === 0) {
+                return;
+            }
+
             nodes = nodesResponse.response;
 
             for (const node of nodes) {
@@ -74,10 +78,4 @@ export class ReviewNodesTask {
             nodes = null;
         }
     }
-
-    private async getEnabledNodes(): Promise<TResult<NodesEntity[]>> {
-        return this.queryBus.execute<GetEnabledNodesQuery, TResult<NodesEntity[]>>(
-            new GetEnabledNodesQuery(),
-        );
-    }
 }

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+export * from './get-enabled-nodes-partial.handler';`
	`2`	`+export * from './get-enabled-nodes-partial.query';`
Original file line number	Diff line number	Diff line change
`@@ -3,5 +3,4 @@ export interface INodeHealthCheckPayload {`
`3`	`3`	`nodeAddress: string;`
`4`	`4`	`nodePort: number \| null;`
`5`	`5`	`isConnected: boolean;`
`6`		`- isConnecting: boolean;`
`7`	`6`	`}`