refactor: consolidate and simplify scheduler tasks

- Remove multiple granular scheduler services
- Streamline task management in scheduler module
- Introduce lightweight tasks for core scheduling functions
- Clean up unused and redundant scheduler-related files
- Update scheduler module to focus on essential periodic jobs

Author: kastov

src/modules/nodes-user-usage-history/builders/bulk-upsert-history-entry/bulk-upsert-history-entry.builder.ts

@@ -22,7 +22,7 @@ export class BulkUpsertHistoryEntryBuilder {
             "total_bytes",
             "created_at",
             "updated_at"
-        ) VALUES ${usageHistoryList.map((usageHistory) => `('${usageHistory.nodeUuid}', '${usageHistory.userUuid}', ${usageHistory.downloadBytes}', '${usageHistory.uploadBytes}', '${usageHistory.totalBytes}', '${date.toISOString()}', NOW())`).join(',')}
+        ) VALUES ${usageHistoryList.map((usageHistory) => `('${usageHistory.nodeUuid}', '${usageHistory.userUuid}', ${usageHistory.downloadBytes}, ${usageHistory.uploadBytes}, ${usageHistory.totalBytes}, '${date.toISOString()}', NOW())`).join(',')}
         ON CONFLICT ("node_uuid","user_uuid","created_at")
         DO UPDATE
             SET

src/modules/nodes/queries/get-all-nodes/get-all-nodes.handler.ts

@@ -0,0 +1,34 @@
+import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
+import { Logger } from '@nestjs/common';
+
+import { ICommandResponse } from '@common/types/command-response.type';
+import { ERRORS } from '@libs/contracts/constants';
+
+import { NodesRepository } from '../../repositories/nodes.repository';
+import { NodesEntity } from '../../entities/nodes.entity';
+import { GetAllNodesQuery } from './get-all-nodes.query';
+
+@QueryHandler(GetAllNodesQuery)
+export class GetAllNodesHandler
+    implements IQueryHandler<GetAllNodesQuery, ICommandResponse<NodesEntity[]>>
+{
+    private readonly logger = new Logger(GetAllNodesHandler.name);
+    constructor(private readonly nodesRepository: NodesRepository) {}
+
+    async execute(): Promise<ICommandResponse<NodesEntity[]>> {
+        try {
+            const nodes = await this.nodesRepository.findAllNodes();
+
+            return {
+                isOk: true,
+                response: nodes,
+            };
+        } catch (error) {
+            this.logger.error(error);
+            return {
+                isOk: false,
+                ...ERRORS.INTERNAL_SERVER_ERROR,
+            };
+        }
+    }
+}

src/modules/nodes/queries/get-all-nodes/get-all-nodes.query.ts

@@ -0,0 +1,3 @@
+export class GetAllNodesQuery {
+    constructor() {}
+}

src/modules/nodes/queries/get-all-nodes/index.ts

@@ -0,0 +1,2 @@
+export * from './get-all-nodes.handler';
+export * from './get-all-nodes.query';

src/modules/nodes/queries/index.ts

@@ -1,5 +1,11 @@
 import { GetNodesByCriteriaHandler } from './get-nodes-by-criteria';
 import { GetEnabledNodesHandler } from './get-enabled-nodes';
 import { GetOnlineNodesHandler } from './get-online-nodes';
+import { GetAllNodesHandler } from './get-all-nodes';
 
-export const QUERIES = [GetEnabledNodesHandler, GetOnlineNodesHandler, GetNodesByCriteriaHandler];
+export const QUERIES = [
+    GetEnabledNodesHandler,
+    GetOnlineNodesHandler,
+    GetNodesByCriteriaHandler,
+    GetAllNodesHandler,
+];

src/modules/nodes/repositories/nodes.repository.ts

@@ -53,6 +53,14 @@ export class NodesRepository implements ICrud<NodesEntity> {
         return nodesList.map((value) => new NodesEntity(value));
     }
 
+    public async findAllNodes(): Promise<NodesEntity[]> {
+        const nodesList = await this.prisma.tx.nodes.findMany({
+            include: ADD_EXCLUSIONS_SELECT,
+        });
+
+        return nodesList.map((value) => new NodesEntity(value));
+    }
+
     public async incrementUsedTraffic(nodeUuid: string, bytes: bigint): Promise<void> {
         await this.prisma.tx.nodes.update({
             where: { uuid: nodeUuid },

src/queue/node-users/node-users.processor.ts

@@ -1,7 +1,6 @@
 import { Job } from 'bullmq';
 
 import { Processor, WorkerHost } from '@nestjs/bullmq';
-import { CommandBus, QueryBus } from '@nestjs/cqrs';
 import { Logger } from '@nestjs/common';
 
 import { AxiosService } from '@common/axios';
@@ -17,11 +16,7 @@ import { QueueNames } from '../queue.enum';
 export class NodeUsersQueueProcessor extends WorkerHost {
     private readonly logger = new Logger(NodeUsersQueueProcessor.name);
 
-    constructor(
-        private readonly queryBus: QueryBus,
-        private readonly commandBus: CommandBus,
-        private readonly axios: AxiosService,
-    ) {
+    constructor(private readonly axios: AxiosService) {
         super();
     }
 

src/queue/queue.enum.ts

@@ -1,7 +1,11 @@
 export enum QueueNames {
     nodeHealthCheck = 'NODE_HEALTH_CHECK_QUEUE',
     nodeUsers = 'NODE_USERS_QUEUE',
+    recordNodeUsage = 'RECORD_NODE_USAGE_QUEUE',
+    recordUserUsage = 'RECORD_USER_USAGE_QUEUE',
+    resetUserTraffic = 'RESET_USER_TRAFFIC_QUEUE',
     startAllNodes = 'START_ALL_NODES_QUEUE',
     startNode = 'START_NODE_QUEUE',
     stopNode = 'STOP_NODE_QUEUE',
+    userJobs = 'USER_JOBS_QUEUE',
 }

src/queue/queue.module.ts

@@ -10,18 +10,26 @@ import { BasicAuthMiddleware } from '@common/middlewares';
 import { useBullBoard } from '@common/utils/startup-app';
 import { BULLBOARD_ROOT } from '@libs/contracts/api';
 
+import { ResetUserTrafficQueueModule } from './reset-user-traffic/reset-user-traffic.module';
 import { NodeHealthCheckQueueModule } from './node-health-check/node-health-check.module';
+import { RecordNodeUsageQueueModule } from './record-node-usage/record-node-usage.module';
+import { RecordUserUsageQueueModule } from './record-user-usage/record-user-usage.module';
 import { StartAllNodesQueueModule } from './start-all-nodes/start-all-nodes.module';
 import { StartNodeQueueModule } from './start-node/start-node.module';
 import { NodeUsersQueueModule } from './node-users/node-users.module';
 import { StopNodeQueueModule } from './stop-node/stop-node.module';
+import { UserJobsQueueModule } from './user-jobs/user-jobs.module';
 
 const queueModules = [
     StartAllNodesQueueModule,
     StartNodeQueueModule,
     StopNodeQueueModule,
     NodeHealthCheckQueueModule,
     NodeUsersQueueModule,
+    RecordNodeUsageQueueModule,
+    RecordUserUsageQueueModule,
+    ResetUserTrafficQueueModule,
+    UserJobsQueueModule,
 ];
 
 const bullBoard = [

src/queue/record-node-usage/enums/index.ts

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