refactor: replace messaging-module with nestjs/microservices

Author: kastov

libs/contract/constants/index.ts

@@ -4,7 +4,6 @@ export * from './errors';
 export * from './events';
 export * from './headers';
 export * from './hosts';
-export * from './messaging';
 export * from './metrics';
 export * from './nodes';
 export * from './oauth2';

libs/contract/constants/messaging/index.ts

@@ -50,7 +50,6 @@
     "@exact-team/telegram-oauth2": "0.0.9",
     "@grammyjs/parse-mode": "1.11.1",
     "@kastov/grammy-nestjs": "0.4.2",
-    "@kastov/messaging-redis-extension": "1.3.1",
     "@kastov/nestjs-prisma-kysely": "0.1.2",
     "@kastov/request-ip": "^0.0.5",
     "@keyv/redis": "^5.1.4",
@@ -65,13 +64,13 @@
     "@nestjs/cqrs": "11.0.3",
     "@nestjs/event-emitter": "3.0.1",
     "@nestjs/jwt": "11.0.1",
+    "@nestjs/microservices": "^11.1.9",
     "@nestjs/passport": "11.0.5",
     "@nestjs/platform-express": "11.1.9",
     "@nestjs/schedule": "6.0.1",
     "@nestjs/serve-static": "5.0.4",
     "@nestjs/swagger": "11.2.3",
     "@nestjs/terminus": "^11.0.0",
-    "@nestjstools/messaging": "^2.22.0",
     "@noble/post-quantum": "^0.5.2",
     "@peculiar/webcrypto": "^1.5.0",
     "@peculiar/x509": "^1.14.2",

libs/contract/constants/messaging/messaging.constants.ts

@@ -5,21 +5,21 @@ import { ClsModule } from 'nestjs-cls';
 import { QueueModule } from 'src/queue/queue.module';
 
 import { TransactionalAdapterPrisma } from '@nestjs-cls/transactional-adapter-prisma';
+import { ConditionalModule, ConfigModule, ConfigService } from '@nestjs/config';
 import { Logger, OnApplicationShutdown, Module } from '@nestjs/common';
 import { ClsPluginTransactional } from '@nestjs-cls/transactional';
-import { ConfigModule, ConfigService } from '@nestjs/config';
 import { EventEmitterModule } from '@nestjs/event-emitter';
 import { CacheModule } from '@nestjs/cache-manager';
 
 import { validateEnvConfig } from '@common/utils/validate-env-config';
 import { PrismaService } from '@common/database/prisma.service';
 import { configSchema, Env } from '@common/config/app-config';
+import { RedisProducerModule } from '@common/microservices';
 import { getRedisConnectionOptions } from '@common/utils';
+import { isProcessor } from '@common/utils/startup-app';
 import { PrismaModule } from '@common/database';
 import { AxiosModule } from '@common/axios';
 
-import { MessagingModules } from '@integration-modules/messaging-modules';
-
 import { RemnawaveModules } from '@modules/remnawave-backend.modules';
 
 @Module({
@@ -75,7 +75,6 @@ import { RemnawaveModules } from '@modules/remnawave-backend.modules';
 
         RemnawaveModules,
         QueueModule,
-        MessagingModules,
         CacheModule.registerAsync({
             imports: [ConfigModule],
             inject: [ConfigService],
@@ -103,6 +102,7 @@ import { RemnawaveModules } from '@modules/remnawave-backend.modules';
                 };
             },
         }),
+        ConditionalModule.registerWhen(RedisProducerModule, () => isProcessor()),
     ],
     controllers: [],
 })

package-lock.json

@@ -19,7 +19,6 @@ import { PrismaModule } from '@common/database';
 import { AxiosModule } from '@common/axios';
 
 import { PrometheusReporterModule } from '@integration-modules/prometheus-reporter/prometheus-reporter.module';
-import { MessagingModules } from '@integration-modules/messaging-modules';
 import { HealthModule } from '@integration-modules/health/health.module';
 
 import { RemnawaveModules } from '@modules/remnawave-backend.modules';
@@ -61,7 +60,6 @@ import { SchedulerModule } from '@scheduler/scheduler.module';
         ScheduleModule.forRoot(),
         SchedulerModule,
         QueueModule,
-        MessagingModules,
         HealthModule,
         CacheModule.registerAsync({
             imports: [ConfigModule],

package.json

@@ -13,10 +13,12 @@ import { json } from 'express';
 import helmet from 'helmet';
 import dayjs from 'dayjs';
 
+import { MicroserviceOptions, Transport } from '@nestjs/microservices';
 import { ConfigService } from '@nestjs/config';
 import { NestFactory } from '@nestjs/core';
 
 import { NotFoundExceptionFilter } from '@common/exception/not-found-exception.filter';
+import { getRedisConnectionOptions } from '@common/utils/get-redis-connection-options';
 import { WorkerRoutesGuard } from '@common/guards/worker-routes/worker-routes.guard';
 import { customLogFilter } from '@common/utils/filter-logs/filter-logs';
 import { isDevelopment } from '@common/utils/startup-app';
@@ -93,6 +95,23 @@ async function bootstrap(): Promise<void> {
         new WorkerRoutesGuard({ allowedPaths: [METRICS_ROOT, BULLBOARD_ROOT, HEALTH_ROOT] }),
     );
 
+    app.connectMicroservice<MicroserviceOptions>({
+        transport: Transport.REDIS,
+        options: {
+            ...getRedisConnectionOptions(
+                config.get<string>('REDIS_SOCKET'),
+                config.get<string>('REDIS_HOST'),
+                config.get<number>('REDIS_PORT'),
+                'ioredis',
+            ),
+            db: config.getOrThrow<number>('REDIS_DB'),
+            password: config.get<string | undefined>('REDIS_PASSWORD'),
+            keyPrefix: 'nmicro:',
+        },
+    });
+
+    await app.startAllMicroservices();
+
     app.enableShutdownHooks();
 
     await app.listen(Number(config.getOrThrow<string>('METRICS_PORT')));

src/bin/processors/processors.root.module.ts

@@ -0,0 +1,2 @@
+export * from './microservices-names.constant';
+export * from './redis-producer.module';

src/bin/scheduler/scheduler.root.module.ts

@@ -0,0 +1,11 @@
+export const MICROSERVICES_NAMES = {
+    REDIS_PRODUCER: 'REDIS_PRODUCER',
+} as const;
+
+export type TMicroservicesNames = typeof MICROSERVICES_NAMES;
+
+export const MESSAGING_NAMES = {
+    NODE_METRICS: 'NODE_METRICS',
+} as const;
+
+export type TMessagingNames = typeof MESSAGING_NAMES;