import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import { HttpService } from '@nestjs/axios'; import { firstValueFrom } from 'rxjs'; interface MensajePendiente { idMensajeWhatsApp: string; idPaquete: string; telefonoDestino: string; tipo: string; mensaje1: string; mensaje2?: string; mensaje3?: string; urlQR?: string; sessionId: string; } @Injectable() export class WorkerService implements OnModuleInit, OnModuleDestroy { private readonly logger = new Logger(WorkerService.name); private isRunning = true; private readonly apiUrl: string; private readonly workerKey: string; private readonly openwaUrl: string; private readonly openwaApiKey: string; private readonly intervalMs: number; // Cache de sesiones OpenWA private sesionesCache: { id: string; name: string; phone: string }[] = []; private sesionesCacheTimestamp = 0; private readonly SESIONES_CACHE_TTL = 5 * 60 * 1000; // 5 minutos constructor( private config: ConfigService, private http: HttpService, ) { this.apiUrl = this.config.get('API_URL') || 'http://localhost:3000/v1'; this.workerKey = this.config.get('WORKER_KEY') || ''; this.openwaUrl = this.config.get('OPENWA_URL') || 'http://localhost:2785'; this.openwaApiKey = this.config.get('OPENWA_API_KEY') || ''; this.intervalMs = Number(this.config.get('WORKER_INTERVAL')) || 30000; } onModuleInit() { this.logger.log(`Worker iniciado. Intervalo entre mensajes: ${this.intervalMs}ms`); this.procesarCiclo(); } onModuleDestroy() { this.isRunning = false; this.logger.log('Worker detenido'); } private timestamp(): string { return new Date().toLocaleTimeString('es-BO', { hour12: false }); } private async procesarCiclo() { if (!this.isRunning) return; await this.procesar(); // Delay entre mensajes const delayMs = this.calcularDelayEntreMensajes(); const delaySeg = (delayMs / 1000).toFixed(1); this.logger.log(`[${this.timestamp()}] ⏳ Esperando ${delaySeg}s antes del siguiente mensaje...`); await this.delay(delayMs); setTimeout(() => this.procesarCiclo(), 0); } private async procesar() { const mensaje = await this.obtenerPendiente(); if (!mensaje) { this.logger.log(`[${this.timestamp()}] No hay mensajes pendientes`); return; } await this.procesarMensaje(mensaje); } private async obtenerPendiente(): Promise { try { const response = await firstValueFrom( this.http.get(`${this.apiUrl}/mensaje/pendientes`, { params: { limit: 1 }, headers: { 'x-worker-key': this.workerKey }, }), ); const data = response.data?.data; return data && data.length > 0 ? data[0] : null; } catch (error: unknown) { const message = error instanceof Error ? error.message : 'Unknown error'; this.logger.error(`[${this.timestamp()}] Error obteniendo mensaje pendiente: ${message}`); return null; } } private async obtenerSesionesDisponibles(): Promise<{ id: string; name: string; phone: string }[]> { const ahora = Date.now(); if (this.sesionesCache.length > 0 && ahora - this.sesionesCacheTimestamp < this.SESIONES_CACHE_TTL) { return this.sesionesCache; } try { const response = await firstValueFrom( this.http.get(`${this.openwaUrl}/api/sessions`, { headers: { 'X-API-Key': this.openwaApiKey }, }), ); const todas = response.data; this.sesionesCache = todas .filter((s: any) => s.status === 'ready') .map((s: any) => ({ id: s.id, name: s.name, phone: s.phone })); this.sesionesCacheTimestamp = ahora; this.logger.log(`[${this.timestamp()}] 🔄 Sesiones OpenWA actualizadas: ${this.sesionesCache.length} disponibles`); } catch (error: unknown) { const message = error instanceof Error ? error.message : 'Unknown error'; this.logger.error(`[${this.timestamp()}] Error obteniendo sesiones OpenWA: ${message}`); } return this.sesionesCache; } private async obtenerSesionRandom(): Promise { const sesiones = await this.obtenerSesionesDisponibles(); if (sesiones.length === 0) { this.logger.warn(`[${this.timestamp()}] ⚠️ No hay sesiones OpenWA disponibles`); return null; } const elegida = sesiones[Math.floor(Math.random() * sesiones.length)]; this.logger.log(`[${this.timestamp()}] 🎲 Sesión seleccionada: ${elegida.name} (${elegida.phone})`); return elegida.id; } private async procesarMensaje(mensaje: MensajePendiente) { const { idMensajeWhatsApp, telefonoDestino, tipo, mensaje1, mensaje2, mensaje3, urlQR } = mensaje; const inicio = Date.now(); this.logger.log(`[${this.timestamp()}] ▶ Procesando mensaje ${idMensajeWhatsApp} → ${telefonoDestino} (${tipo})`); try { const sessionId = await this.obtenerSesionRandom(); if (!sessionId) { await this.actualizarEstado(idMensajeWhatsApp, 'ERROR', 'No hay sesiones OpenWA disponibles'); return; } const chatId = `${telefonoDestino}@c.us`; const mensajes: string[] = [mensaje1, mensaje2, mensaje3].filter((m): m is string => Boolean(m)); // ── PASO 1: Enviar mensaje1 + QR INMEDIATO (sin delay) ── if (mensajes.length > 0) { const texto1 = mensajes[0]; if (tipo === 'RECEPCION_PAQUETE' && urlQR) { // Intentar enviar imagen; si falla, enviar QR como link try { await this.enviarImagen(sessionId, chatId, urlQR, texto1); this.logger.log(`[${this.timestamp()}] 📷✉️ Mensaje 1 (QR + imagen) enviado INMEDIATO`); } catch { this.logger.warn(`[${this.timestamp()}] ⚠️ Error enviando imagen, usando fallback con link`); const fullUrl = urlQR.startsWith('http') ? urlQR : `${this.apiUrl.replace('/v1', '')}${urlQR}`; const textoConQR = `${texto1}\n\n🔗 *Ver QR:* ${fullUrl}`; await this.enviarTexto(sessionId, chatId, textoConQR); this.logger.log(`[${this.timestamp()}] ✉️ Mensaje 1 (QR como link) enviado INMEDIATO`); } // Cambiar estado a ENVIANDO después del primer mensaje (RECEPCION_PAQUETE tiene 3 mensajes) if (mensajes.length > 1) { await this.actualizarEstado(idMensajeWhatsApp, 'ENVIANDO'); this.logger.log(`[${this.timestamp()}] 🔄 Estado cambiado a ENVIANDO`); } } else { await this.enviarTexto(sessionId, chatId, texto1); this.logger.log(`[${this.timestamp()}] ✉️ Mensaje 1 enviado INMEDIATO`); } } // ── PASO 2: Calcular y guardar delay1 para usarlo al final ── const texto1ParaDelay = mensajes[0] || ''; const delay1 = this.calcularDelayEscritura(texto1ParaDelay.length); const delay1Seg = (delay1 / 1000).toFixed(1); this.logger.log(`[${this.timestamp()}] ⏳ Delay1 calculado: ${delay1Seg}s (para usar después)`); // ── PASOS 3-6: Enviar mensajes 2 y 3 con delays y typing ── for (let i = 1; i < mensajes.length; i++) { const texto = mensajes[i]; // Calcular delay para este mensaje const delayMsg = this.calcularDelayEscritura(texto.length); const delayMsgSeg = (delayMsg / 1000).toFixed(1); // Typing indicator + esperar delay await this.setTyping(sessionId, chatId); this.logger.log(`[${this.timestamp()}] ⌨️ Typing mensaje ${i + 1}...`); this.logger.log(`[${this.timestamp()}] ⏳ Esperando ${delayMsgSeg}s (${texto.length} chars)`); await this.delay(delayMsg); // Enviar mensaje await this.enviarTexto(sessionId, chatId, texto); this.logger.log(`[${this.timestamp()}] ✉️ Mensaje ${i + 1} enviado`); // Delay entre mensajes (si hay más) if (i < mensajes.length - 1) { const delayEntre = this.calcularDelayEntreMensajes(); const delayEntreSeg = (delayEntre / 1000).toFixed(1); this.logger.log(`[${this.timestamp()}] ⏳ Esperando ${delayEntreSeg}s antes del siguiente mensaje...`); await this.delay(delayEntre); } } // Marcar como enviado ANTES del delay para no perder el estado si el worker se cae await this.actualizarEstado(idMensajeWhatsApp, 'ENVIADO'); const totalSeg = ((Date.now() - inicio) / 1000).toFixed(1); this.logger.log(`[${this.timestamp()}] ✅ Mensaje ${idMensajeWhatsApp} ENVIADO (${mensajes.length} mensajes) → tiempo total: ${totalSeg}s`); // ── PASO 7: Esperar delay1 (el que se calculó en paso 2) ── this.logger.log(`[${this.timestamp()}] ⏳ Esperando delay1: ${delay1Seg}s antes de continuar...`); await this.delay(delay1); } catch (error: any) { const message = error instanceof Error ? error.message : 'Unknown error'; const responseMessage = error?.response?.data?.message || error?.response?.data?.error || ''; const fullError = responseMessage ? `${message} → ${responseMessage}` : message; await this.actualizarEstado(idMensajeWhatsApp, 'ERROR', fullError); const totalSeg = ((Date.now() - inicio) / 1000).toFixed(1); this.logger.error(`[${this.timestamp()}] ❌ Error en ${idMensajeWhatsApp}: ${fullError} (${totalSeg}s)`); } } private calcularDelayEscritura(caracteres: number): number { const caracteresPorMinuto = 350; const base = (caracteres / caracteresPorMinuto) * 60 * 1000; const random = 1 + (Math.random() * 0.6 - 0.3); const reaccion = 500 + Math.random() * 1000; const delay = base * random + reaccion; return Math.min(Math.max(delay, 3000), 120000); // 3s-2min } private calcularDelayEntreMensajes(): number { const base = 3000; // 3s base entre mensajes const random = 1 + (Math.random() * 0.6 - 0.3); // ±30% return Math.min(Math.max(base * random, 2000), 8000); // 2s-8s } private delay(ms: number): Promise { return new Promise(resolve => setTimeout(resolve, ms)); } private async setTyping(sessionId: string, chatId: string) { try { await firstValueFrom( this.http.post( `${this.openwaUrl}/api/sessions/${sessionId}/chat-state`, { chatId, state: 'composing' }, { headers: { 'X-API-Key': this.openwaApiKey, 'Content-Type': 'application/json', }, }, ), ); } catch { // Silenciar errores de typing — no es crítico } } private async enviarTexto(sessionId: string, chatId: string, text: string) { const response = await firstValueFrom( this.http.post( `${this.openwaUrl}/api/sessions/${sessionId}/messages/send-text`, { chatId, text }, { headers: { 'X-API-Key': this.openwaApiKey, 'Content-Type': 'application/json', }, }, ), ); return response.data; } private async enviarImagen(sessionId: string, chatId: string, imageUrl: string, caption: string) { const fullUrl = imageUrl.startsWith('http') ? imageUrl : `http://localhost:3000${imageUrl}`; const response = await firstValueFrom(this.http.get(fullUrl, { responseType: 'arraybuffer' })); const base64Data = Buffer.from(response.data).toString('base64'); const dataUrl = `data:image/png;base64,${base64Data}`; const result = await firstValueFrom( this.http.post( `${this.openwaUrl}/api/sessions/${sessionId}/messages/send-image`, { chatId, base64: dataUrl, caption, mimetype: 'image/png', filename: 'qr.png', }, { headers: { 'X-API-Key': this.openwaApiKey, 'Content-Type': 'application/json', }, }, ), ); return result.data; } private async actualizarEstado(id: string, estado: string, error?: string) { try { await firstValueFrom( this.http.put( `${this.apiUrl}/mensaje/${id}/estado`, { estado, error: error || null, fechaEnvio: new Date().toISOString(), }, { headers: { 'x-worker-key': this.workerKey }, }, ), ); } catch (err: unknown) { const message = err instanceof Error ? err.message : 'Unknown error'; this.logger.error(`[${this.timestamp()}] Error actualizando estado del mensaje ${id}: ${message}`); } } }