feat: Add NATS integration support to the event system

- Added NATS package to dependencies
- Created Prisma schema models for NATS configuration
- Implemented NATS controller, router, and event management
- Updated instance controller and event manager to support NATS
- Added NATS configuration options in environment configuration
- Included NATS events in instance validation schema

src/api/integrations/event/event.dto.ts

@@ -26,6 +26,11 @@ export class EventDto {
     events?: string[];
   };
 
+  nats?: {
+    enabled?: boolean;
+    events?: string[];
+  };
+
   pusher?: {
     enabled?: boolean;
     appId?: string;
@@ -63,6 +68,11 @@ export function EventInstanceMixin<TBase extends Constructor>(Base: TBase) {
       events?: string[];
     };
 
+    nats?: {
+      enabled?: boolean;
+      events?: string[];
+    };
+
     pusher?: {
       enabled?: boolean;
       appId?: string;

src/api/integrations/event/event.manager.ts

@@ -1,3 +1,4 @@
+import { NatsController } from '@api/integrations/event/nats/nats.controller';
 import { PusherController } from '@api/integrations/event/pusher/pusher.controller';
 import { RabbitmqController } from '@api/integrations/event/rabbitmq/rabbitmq.controller';
 import { SqsController } from '@api/integrations/event/sqs/sqs.controller';
@@ -13,6 +14,7 @@ export class EventManager {
   private websocketController: WebsocketController;
   private webhookController: WebhookController;
   private rabbitmqController: RabbitmqController;
+  private natsController: NatsController;
   private sqsController: SqsController;
   private pusherController: PusherController;
 
@@ -23,6 +25,7 @@ export class EventManager {
     this.websocket = new WebsocketController(prismaRepository, waMonitor);
     this.webhook = new WebhookController(prismaRepository, waMonitor);
     this.rabbitmq = new RabbitmqController(prismaRepository, waMonitor);
+    this.nats = new NatsController(prismaRepository, waMonitor);
     this.sqs = new SqsController(prismaRepository, waMonitor);
     this.pusher = new PusherController(prismaRepository, waMonitor);
   }
@@ -67,6 +70,14 @@ export class EventManager {
     return this.rabbitmqController;
   }
 
+  public set nats(nats: NatsController) {
+    this.natsController = nats;
+  }
+
+  public get nats() {
+    return this.natsController;
+  }
+
   public set sqs(sqs: SqsController) {
     this.sqsController = sqs;
   }
@@ -85,6 +96,7 @@ export class EventManager {
   public init(httpServer: Server): void {
     this.websocket.init(httpServer);
     this.rabbitmq.init();
+    this.nats.init();
     this.sqs.init();
     this.pusher.init();
   }
@@ -103,6 +115,7 @@ export class EventManager {
   }): Promise<void> {
     await this.websocket.emit(eventData);
     await this.rabbitmq.emit(eventData);
+    await this.nats.emit(eventData);
     await this.sqs.emit(eventData);
     await this.webhook.emit(eventData);
     await this.pusher.emit(eventData);
@@ -125,6 +138,14 @@ export class EventManager {
         },
       });
 
+    if (data.nats)
+      await this.nats.set(instanceName, {
+        nats: {
+          enabled: true,
+          events: data.nats?.events,
+        },
+      });
+
     if (data.sqs)
       await this.sqs.set(instanceName, {
         sqs: {

src/api/integrations/event/event.router.ts

@@ -1,3 +1,4 @@
+import { NatsRouter } from '@api/integrations/event/nats/nats.router';
 import { PusherRouter } from '@api/integrations/event/pusher/pusher.router';
 import { RabbitmqRouter } from '@api/integrations/event/rabbitmq/rabbitmq.router';
 import { SqsRouter } from '@api/integrations/event/sqs/sqs.router';
@@ -14,6 +15,7 @@ export class EventRouter {
     this.router.use('/webhook', new WebhookRouter(configService, ...guards).router);
     this.router.use('/websocket', new WebsocketRouter(...guards).router);
     this.router.use('/rabbitmq', new RabbitmqRouter(...guards).router);
+    this.router.use('/nats', new NatsRouter(...guards).router);
     this.router.use('/pusher', new PusherRouter(...guards).router);
     this.router.use('/sqs', new SqsRouter(...guards).router);
   }

src/api/integrations/event/event.schema.ts

@@ -16,6 +16,9 @@ export const eventSchema: JSONSchema7 = {
     rabbitmq: {
       $ref: '#/$defs/event',
     },
+    nats: {
+      $ref: '#/$defs/event',
+    },
     sqs: {
       $ref: '#/$defs/event',
     },

src/api/integrations/event/nats/nats.controller.ts

@@ -0,0 +1,161 @@
+import { PrismaRepository } from '@api/repository/repository.service';
+import { WAMonitoringService } from '@api/services/monitor.service';
+import { configService, Log, Nats } from '@config/env.config';
+import { Logger } from '@config/logger.config';
+import { connect, NatsConnection, StringCodec } from 'nats';
+
+import { EmitData, EventController, EventControllerInterface } from '../event.controller';
+
+export class NatsController extends EventController implements EventControllerInterface {
+  public natsClient: NatsConnection | null = null;
+  private readonly logger = new Logger('NatsController');
+  private readonly sc = StringCodec();
+
+  constructor(prismaRepository: PrismaRepository, waMonitor: WAMonitoringService) {
+    super(prismaRepository, waMonitor, configService.get<Nats>('NATS')?.ENABLED, 'nats');
+  }
+
+  public async init(): Promise<void> {
+    if (!this.status) {
+      return;
+    }
+
+    try {
+      const uri = configService.get<Nats>('NATS').URI;
+
+      this.natsClient = await connect({ servers: uri });
+
+      this.logger.info('NATS initialized');
+
+      if (configService.get<Nats>('NATS')?.GLOBAL_ENABLED) {
+        await this.initGlobalSubscriptions();
+      }
+    } catch (error) {
+      this.logger.error('Failed to connect to NATS:');
+      this.logger.error(error);
+      throw error;
+    }
+  }
+
+  public async emit({
+    instanceName,
+    origin,
+    event,
+    data,
+    serverUrl,
+    dateTime,
+    sender,
+    apiKey,
+    integration,
+  }: EmitData): Promise<void> {
+    if (integration && !integration.includes('nats')) {
+      return;
+    }
+
+    if (!this.status || !this.natsClient) {
+      return;
+    }
+
+    const instanceNats = await this.get(instanceName);
+    const natsLocal = instanceNats?.events;
+    const natsGlobal = configService.get<Nats>('NATS').GLOBAL_ENABLED;
+    const natsEvents = configService.get<Nats>('NATS').EVENTS;
+    const prefixKey = configService.get<Nats>('NATS').PREFIX_KEY;
+    const we = event.replace(/[.-]/gm, '_').toUpperCase();
+    const logEnabled = configService.get<Log>('LOG').LEVEL.includes('WEBHOOKS');
+
+    const message = {
+      event,
+      instance: instanceName,
+      data,
+      server_url: serverUrl,
+      date_time: dateTime,
+      sender,
+      apikey: apiKey,
+    };
+
+    // Instância específica
+    if (instanceNats?.enabled) {
+      if (Array.isArray(natsLocal) && natsLocal.includes(we)) {
+        const subject = `${instanceName}.${event.toLowerCase()}`;
+
+        try {
+          this.natsClient.publish(subject, this.sc.encode(JSON.stringify(message)));
+
+          if (logEnabled) {
+            const logData = {
+              local: `${origin}.sendData-NATS`,
+              ...message,
+            };
+            this.logger.log(logData);
+          }
+        } catch (error) {
+          this.logger.error(`Failed to publish to NATS (instance): ${error}`);
+        }
+      }
+    }
+
+    // Global
+    if (natsGlobal && natsEvents[we]) {
+      try {
+        const subject = prefixKey ? `${prefixKey}.${event.toLowerCase()}` : event.toLowerCase();
+
+        this.natsClient.publish(subject, this.sc.encode(JSON.stringify(message)));
+
+        if (logEnabled) {
+          const logData = {
+            local: `${origin}.sendData-NATS-Global`,
+            ...message,
+          };
+          this.logger.log(logData);
+        }
+      } catch (error) {
+        this.logger.error(`Failed to publish to NATS (global): ${error}`);
+      }
+    }
+  }
+
+  private async initGlobalSubscriptions(): Promise<void> {
+    this.logger.info('Initializing global subscriptions');
+
+    const events = configService.get<Nats>('NATS').EVENTS;
+    const prefixKey = configService.get<Nats>('NATS').PREFIX_KEY;
+
+    if (!events) {
+      this.logger.warn('No events to initialize on NATS');
+      return;
+    }
+
+    const eventKeys = Object.keys(events);
+
+    for (const event of eventKeys) {
+      if (events[event] === false) continue;
+
+      const subject = prefixKey ? `${prefixKey}.${event.toLowerCase()}` : event.toLowerCase();
+
+      // Criar uma subscription para cada evento
+      try {
+        const subscription = this.natsClient.subscribe(subject);
+        this.logger.info(`Subscribed to: ${subject}`);
+
+        // Processar mensagens (exemplo básico)
+        (async () => {
+          for await (const msg of subscription) {
+            try {
+              const data = JSON.parse(this.sc.decode(msg.data));
+              // Aqui você pode adicionar a lógica de processamento
+              this.logger.debug(`Received message on ${subject}:`);
+              this.logger.debug(data);
+            } catch (error) {
+              this.logger.error(`Error processing message on ${subject}:`);
+              this.logger.error(error);
+            }
+          }
+        })();
+      } catch (error) {
+        this.logger.error(`Failed to subscribe to ${subject}:`);
+        this.logger.error(error);
+      }
+    }
+  }
+}

src/api/integrations/event/nats/nats.router.ts

@@ -0,0 +1,36 @@
+import { RouterBroker } from '@api/abstract/abstract.router';
+import { InstanceDto } from '@api/dto/instance.dto';
+import { EventDto } from '@api/integrations/event/event.dto';
+import { HttpStatus } from '@api/routes/index.router';
+import { eventManager } from '@api/server.module';
+import { eventSchema, instanceSchema } from '@validate/validate.schema';
+import { RequestHandler, Router } from 'express';
+
+export class NatsRouter extends RouterBroker {
+  constructor(...guards: RequestHandler[]) {
+    super();
+    this.router
+      .post(this.routerPath('set'), ...guards, async (req, res) => {
+        const response = await this.dataValidate<EventDto>({
+          request: req,
+          schema: eventSchema,
+          ClassRef: EventDto,
+          execute: (instance, data) => eventManager.nats.set(instance.instanceName, data),
+        });
+
+        res.status(HttpStatus.CREATED).json(response);
+      })
+      .get(this.routerPath('find'), ...guards, async (req, res) => {
+        const response = await this.dataValidate<InstanceDto>({
+          request: req,
+          schema: instanceSchema,
+          ClassRef: InstanceDto,
+          execute: (instance) => eventManager.nats.get(instance.instanceName),
+        });
+
+        res.status(HttpStatus.OK).json(response);
+      });
+  }
+
+  public readonly router: Router = Router();
+}