Hugging Face's logo

Spaces:
oex2003
/
evolution-api
Runtime error

App Files Community
evolution-api
File size: 4,901 Bytes 
5e518ea
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
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);
      }
    }
  }
}