VPS 4 Deploy Agent
eca65bf765
Secretária IA mais segura e contextual: - Roteamento por especialidade: /contexto (dentistas+especialidades) injetado no cérebro; descobrir o motivo e priorizar clínico geral só p/ avaliação/limpeza. - Situações da clínica: regras editáveis injetadas no prompt (siga à risca). - Dúvida/caso incerto → tool encaminhar_humano (não agenda; pede p/ aguardar). - Imagem/áudio: ext:message.transcription religa no fluxo ext:message.new (guard hadCaption evita processar 2x); a IA reage à descrição da imagem. - Checklist do dono × banco → cautelas no /contexto: NÃO assumir clínico geral por padrão p/ quem já é paciente; sem certeza de quem/qual dentista → humano. - Desligar GLOBAL da Secretária: GET/POST /secretaria/power (kill-switch auto_reply via config.set, efeito imediato, sem restart; visível a todos). Inclui a plumbing da sessão (chat.upsert, ownership de conexão, ws-bridge). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1387 lines
66 KiB
TypeScript
1387 lines
66 KiB
TypeScript
/**
|
||
* WhatsAppConnectionManager — Orquestra conexões Baileys para todas as instâncias.
|
||
*
|
||
* Esta é a classe central do módulo WhatsApp. Ela gerencia:
|
||
* 1. Ciclo de vida das conexões (connect, disconnect, reconnect)
|
||
* 2. Registro de TODOS os event listeners do Baileys
|
||
* 3. Reconexão automática com backoff exponencial
|
||
* 4. Watchdog para detectar conexões mortas silenciosamente
|
||
* 5. Sincronização de chats, contatos e histórico
|
||
* 6. Mapeamento LID → phone JID (via chats.phoneNumberShare)
|
||
* 7. Emissão de eventos Socket.IO para o frontend
|
||
*
|
||
* ARQUITETURA:
|
||
* - Uma instância de WhatsAppConnectionManager por servidor (singleton prático)
|
||
* - Cada instância WhatsApp (número conectado) tem seu próprio WASocket
|
||
* - Os Maps internos indexam por instanceId para acesso O(1)
|
||
* - Multi-tenant: tenantByInstance mapeia instância → tenant para isolamento
|
||
*
|
||
* EVENTOS BAILEYS REGISTRADOS:
|
||
* - creds.update → salva credenciais (auth state)
|
||
* - connection.update → gerencia QR, conexão, desconexão, reconexão
|
||
* - messages.upsert → delega para MessageHandler (tempo real + histórico)
|
||
* - messages.update → atualiza status (sent→delivered→read)
|
||
* - contacts.upsert/update → delega para ContactHandler
|
||
* - chats.upsert/update → sincroniza lista de chats no banco
|
||
* - chats.phoneNumberShare → captura mapeamento LID → phone JID
|
||
* - groups.upsert/update → sincroniza grupos (subject, metadata)
|
||
* - messaging-history.set → history sync (contatos, chats, mensagens em lote)
|
||
*
|
||
* @see MessageHandler — processa mensagens individuais
|
||
* @see ContactHandler — gerencia contatos e avatares
|
||
*/
|
||
import path from 'path'
|
||
import fs from 'fs/promises'
|
||
import dns from 'dns/promises'
|
||
import {
|
||
getEngine,
|
||
DisconnectReason,
|
||
type WASocket,
|
||
type ConnectionState,
|
||
} from '../engine'
|
||
import { Boom } from '@hapi/boom'
|
||
import QRCode from 'qrcode'
|
||
import { env } from '../../../config/env'
|
||
import { logger } from '../../../config/logger'
|
||
import { dragonfly } from '../../../infra/cache/dragonfly'
|
||
import { prisma } from '../../../infra/database/prisma'
|
||
import { InstanceStatus } from '@prisma/client'
|
||
import { MessageHandler } from '../handlers/MessageHandler'
|
||
import { ContactHandler } from '../handlers/ContactHandler'
|
||
import { ChatbotService } from '../../chatbot/chatbot.service'
|
||
import type { Server as SocketIOServer } from 'socket.io'
|
||
import { hookBus } from '../../../core/hook-bus'
|
||
import { invalidateChatListCache } from "../../chats/chat.cache"
|
||
|
||
// ─── Chaves de cache no DragonflyDB (Redis-compatible) ──────────────────────
|
||
const CACHE_KEY_STATUS = (instanceId: string) => `instance:${instanceId}:status`
|
||
const CACHE_KEY_QR = (instanceId: string) => `instance:${instanceId}:qr`
|
||
|
||
// ─── Configuração de reconexão automática ───────────────────────────────────
|
||
// Backoff exponencial: 3s → 6s → 12s → 24s → 48s → ... → max 5min
|
||
// Nunca desiste, exceto loggedOut (usuário deslogou) ou forbidden (ban)
|
||
const RECONNECT_BASE_DELAY_MS = 3_000 // delay inicial
|
||
const RECONNECT_MAX_DELAY_MS = 300_000 // teto geral: 5 minutos
|
||
// Teto MENOR para erros transientes de rede (DNS, ECONN, ETIMEDOUT).
|
||
// Esses erros costumam resolver em segundos; um teto de 5min deixa a
|
||
// sessão "dormindo" num setTimeout longo enquanto a rede já voltou.
|
||
const RECONNECT_NETWORK_MAX_MS = 60_000 // teto rede: 60s
|
||
const RECONNECT_MAX_ATTEMPTS = Infinity
|
||
|
||
// ─── Network probe — acorda reconexões pendentes quando a rede volta ────────
|
||
// Enquanto uma instância está aguardando backoff, um DNS lookup leve é
|
||
// disparado em intervalo fixo. Se o lookup funciona E existe um retry
|
||
// agendado para daqui a mais de PROBE_ADVANCE_THRESHOLD_MS, cancela o timer
|
||
// e dispara retry imediato. Zero tráfego para o WhatsApp — só DNS.
|
||
const NETWORK_PROBE_INTERVAL_MS = 30_000 // a cada 30s
|
||
const PROBE_ADVANCE_THRESHOLD_MS = 15_000 // só adianta se faltar > 15s
|
||
const PROBE_DNS_HOST = 'web.whatsapp.com'
|
||
|
||
// ─── Watchdog: detecta conexão morta silenciosamente ────────────────────────
|
||
// Conservador por design: reconexões frequentes sinalizam "cliente instável"
|
||
// para o WhatsApp e aumentam risco de banimento do número. Por isso:
|
||
// - Timeout alto (10 min): conversas calmas são normais, não disparam
|
||
// - Intervalo baixo de verificação (3 min): menos overhead
|
||
// - Ping leve antes de matar: tenta reviver sem derrubar o socket
|
||
// - Cool-down entre kills: evita loop de reconexão se a rede oscila
|
||
const WATCHDOG_INTERVAL_MS = 180_000 // verifica a cada 3 min
|
||
const WATCHDOG_TIMEOUT_MS = 600_000 // 10 min sem atividade antes de agir
|
||
const WATCHDOG_PING_GRACE_MS = 30_000 // grace padrão (official): 30s
|
||
const WATCHDOG_PING_GRACE_INFIN_MS = 60_000 // grace InfiniteAPI: 60s (ACK mais lento)
|
||
const WATCHDOG_COOLDOWN_MS = 900_000 // 15 min entre force-kills consecutivos
|
||
|
||
export class WhatsAppConnectionManager {
|
||
// ─── Maps indexados por instanceId (UUID) ──────────────────────────────
|
||
/** Socket Baileys ativo por instância */
|
||
private sockets = new Map<string, WASocket>()
|
||
/** Handler de contatos por instância */
|
||
private contactHandlers = new Map<string, ContactHandler>()
|
||
/** Handler de mensagens por instância */
|
||
private msgHandlers = new Map<string, MessageHandler>()
|
||
/** Contador de tentativas de reconexão (reseta ao conectar com sucesso) */
|
||
private reconnectAttempts = new Map<string, number>()
|
||
/** Timer de reconexão pendente (clearTimeout para cancelar) */
|
||
private reconnectTimers = new Map<string, ReturnType<typeof setTimeout>>()
|
||
/**
|
||
* Timestamp absoluto (ms) em que o retry pendente vai disparar.
|
||
* Usado pelo network probe para decidir se vale adiantar um retry
|
||
* que ainda está dormindo num backoff longo.
|
||
*/
|
||
private reconnectScheduledAt = new Map<string, number>()
|
||
/** Timer global do network probe (singleton) */
|
||
private networkProbeTimer: ReturnType<typeof setInterval> | null = null
|
||
/** Timer do watchdog (clearInterval para parar) */
|
||
private watchdogTimers = new Map<string, ReturnType<typeof setInterval>>()
|
||
/** Timestamp da última atividade (qualquer evento Baileys) */
|
||
private lastActivity = new Map<string, number>()
|
||
/** Timestamp do último force-kill do watchdog (cool-down anti-loop) */
|
||
private lastWatchdogKill = new Map<string, number>()
|
||
/** Timestamp do último ping do watchdog (aguarda resposta antes de matar) */
|
||
private lastWatchdogPing = new Map<string, number>()
|
||
/** Mapeamento instanceId → tenantId (necessário para reconexões) */
|
||
private tenantByInstance = new Map<string, string>()
|
||
/** Engine ativa por instância (usada pelo watchdog para ajustar tolerâncias) */
|
||
private engineByInstance = new Map<string, string>()
|
||
|
||
/** Socket.IO server compartilhado — emite eventos para o frontend */
|
||
private io: SocketIOServer
|
||
/** Serviço de chatbot — processa mensagens recebidas com IA */
|
||
private chatbotService: ChatbotService
|
||
|
||
constructor(io: SocketIOServer) {
|
||
this.io = io
|
||
this.chatbotService = new ChatbotService(io)
|
||
this.startNetworkProbe()
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// CONEXÃO PRINCIPAL
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Conecta (ou reconecta) uma instância WhatsApp.
|
||
*
|
||
* Fluxo:
|
||
* 1. Carrega auth state do disco (useMultiFileAuthState)
|
||
* 2. Cria socket Baileys com configuração padrão
|
||
* 3. Registra TODOS os event listeners
|
||
* 4. Baileys inicia handshake com WhatsApp Web
|
||
* 5. Se precisar de QR: emite evento 'qr' via Socket.IO
|
||
* 6. Se já autenticado: connection 'open' dispara sync
|
||
*
|
||
* IMPORTANTE: Este método é idempotente — se a instância já está
|
||
* conectada, retorna sem fazer nada (evita sockets duplicados).
|
||
*/
|
||
async connect(instanceId: string, tenantId: string): Promise<void> {
|
||
if (this.sockets.has(instanceId)) {
|
||
// Socket já existe — re-emite o status atual para sincronizar o frontend
|
||
// (acontece quando o backend reconectou em background antes do frontend abrir)
|
||
const cachedStatus = await dragonfly.get(CACHE_KEY_STATUS(instanceId)) as string | null
|
||
const currentStatus = cachedStatus ?? InstanceStatus.CONNECTED
|
||
logger.info({ instanceId, currentStatus }, 'Instância já conectada — re-emitindo status para o frontend')
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:status', {
|
||
instanceId,
|
||
status: currentStatus,
|
||
})
|
||
return
|
||
}
|
||
|
||
// Limpa timers pendentes de reconexão anterior
|
||
this.clearReconnectTimer(instanceId)
|
||
this.tenantByInstance.set(instanceId, tenantId)
|
||
|
||
// ── Auth state: credenciais persistidas em disco ──────────────────────
|
||
// Cada instância tem seu diretório em BAILEYS_SESSIONS_PATH/{instanceId}/
|
||
// Contém: creds.json, app-state-sync-key-*, pre-key-*, etc.
|
||
const sessionDir = path.resolve(env.BAILEYS_SESSIONS_PATH, instanceId)
|
||
|
||
try {
|
||
await fs.mkdir(sessionDir, { recursive: true })
|
||
} catch (err) {
|
||
logger.error({ err, instanceId, sessionDir }, 'Falha ao criar diretório de sessão')
|
||
await this.updateInstanceStatus(instanceId, InstanceStatus.DISCONNECTED)
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:status', { instanceId, status: InstanceStatus.DISCONNECTED })
|
||
return
|
||
}
|
||
|
||
// ── Seleciona engine por instância ───────────────────────────────────
|
||
// Lê o campo engine da instância no banco (padrão: "infinite").
|
||
// Permite que cada instância use uma engine diferente sem reiniciar o container.
|
||
const instanceRecord = await prisma.instance.findFirst({
|
||
where: { id: instanceId },
|
||
select: { engine: true },
|
||
})
|
||
const engineType = (instanceRecord as any)?.engine ?? 'infinite'
|
||
const {
|
||
makeWASocket,
|
||
useMultiFileAuthState,
|
||
fetchLatestBaileysVersion,
|
||
makeCacheableSignalKeyStore,
|
||
} = getEngine(engineType)
|
||
this.engineByInstance.set(instanceId, engineType)
|
||
logger.info({ instanceId, engine: engineType }, 'Engine selecionada para instância')
|
||
|
||
const { state, saveCreds } = await useMultiFileAuthState(sessionDir)
|
||
|
||
// fetchLatestBaileysVersion faz requisição HTTP ao WhatsApp — pode falhar
|
||
// se não houver internet. Usa versão embutida como fallback seguro.
|
||
let version: [number, number, number] = [2, 3000, 1023504162]
|
||
try {
|
||
const fetched = await fetchLatestBaileysVersion()
|
||
version = fetched.version
|
||
} catch (err) {
|
||
logger.warn({ instanceId }, 'fetchLatestBaileysVersion falhou — usando versão embutida como fallback')
|
||
}
|
||
|
||
logger.info({ instanceId, version }, 'Iniciando socket Baileys')
|
||
|
||
// ── Cria socket Baileys ──────────────────────────────────────────────
|
||
// DEBUG: logger trace só para capturar XML de mensagens enviadas
|
||
const debugLogger: any = {
|
||
level: 'trace',
|
||
child: () => debugLogger,
|
||
trace: (obj: any) => {
|
||
if (obj?.xml && typeof obj.xml === 'string' && obj.xml.startsWith('<message')) {
|
||
console.log('[NEWWHATS/STANZA-XML]\n' + obj.xml)
|
||
}
|
||
},
|
||
debug: () => {}, info: () => {}, warn: () => {}, error: () => {}, fatal: () => {},
|
||
}
|
||
|
||
const sock = makeWASocket({
|
||
version,
|
||
auth: {
|
||
creds: state.creds,
|
||
// Cache de chaves Signal para performance (evita leituras de disco)
|
||
keys: makeCacheableSignalKeyStore(state.keys, logger as any),
|
||
},
|
||
printQRInTerminal: false,
|
||
qrTimeout: 60_000,
|
||
logger: debugLogger,
|
||
browser: ['NewWhats', 'Chrome', '124.0.0'],
|
||
syncFullHistory: true, // Solicita histórico completo ao telefone
|
||
markOnlineOnConnect: true, // Marca como "online" ao conectar
|
||
// getMessage: callback para reenvio de mensagens (retry)
|
||
// O Baileys chama quando precisa reenviar uma mensagem que falhou
|
||
getMessage: async (key) => {
|
||
if (!key.id) return undefined
|
||
const msg = await prisma.message.findUnique({
|
||
where: { instanceId_messageId: { instanceId, messageId: key.id } },
|
||
})
|
||
if (!msg?.body) return undefined
|
||
return { conversation: msg.body }
|
||
},
|
||
})
|
||
|
||
// ── Kill switch de saída (modo teste) ──────────────────────────────────────
|
||
// DISABLE_OUTBOUND=true impede QUALQUER envio ao WhatsApp nesta sessão.
|
||
// Como todo envio (rotas, IA/secretária, chatbot, scheduler, warmer, plugins)
|
||
// usa este sock, envolver sendMessage aqui é o backstop universal para os
|
||
// testes: nada é disparado para contatos/grupos reais.
|
||
if (process.env.DISABLE_OUTBOUND === 'true') {
|
||
sock.sendMessage = (async (jid: string) => {
|
||
logger.warn({ instanceId, jid },
|
||
'[OUTBOUND BLOQUEADO] DISABLE_OUTBOUND ativo — mensagem NÃO enviada (modo teste)')
|
||
return { key: { id: `BLOCKED_${Date.now()}`, remoteJid: jid, fromMe: true }, status: 1 } as any
|
||
}) as typeof sock.sendMessage
|
||
logger.warn({ instanceId },
|
||
'[OUTBOUND BLOQUEADO] Modo teste ativo — envios desativados nesta sessão')
|
||
}
|
||
|
||
this.sockets.set(instanceId, sock)
|
||
this.touchActivity(instanceId)
|
||
|
||
// ═════════════════════════════════════════════════════════════════════
|
||
// REGISTRO DE EVENT LISTENERS DO BAILEYS
|
||
// ═════════════════════════════════════════════════════════════════════
|
||
|
||
// ─── Credenciais: salvar ao atualizar ────────────────────────────────
|
||
// O Baileys atualiza credenciais periodicamente (rotação de chaves Signal)
|
||
;(sock.ev as any).on('creds.update', saveCreds)
|
||
|
||
// ─── Estado de conexão ───────────────────────────────────────────────
|
||
// Eventos: qr (novo QR gerado), open (conectado), close (desconectado),
|
||
// connecting (tentando conectar)
|
||
;(sock.ev as any).on('connection.update', (update: any) => {
|
||
this.touchActivity(instanceId)
|
||
this.baileysLog(tenantId, instanceId, 'connection.update', {
|
||
connection: update.connection,
|
||
qr: update.qr ? '(qr gerado)' : undefined,
|
||
lastDisconnect: update.lastDisconnect
|
||
? { statusCode: (update.lastDisconnect.error as any)?.output?.statusCode }
|
||
: undefined,
|
||
})
|
||
this.handleConnectionUpdate(instanceId, tenantId, update, sock)
|
||
})
|
||
|
||
// ─── Mensagens recebidas (tempo real + histórico) ────────────────────
|
||
// type 'notify': mensagem em tempo real → processMessage completo
|
||
// type 'append': histórico sync → handleHistory (bulk, sem socket)
|
||
const msgHandler = new MessageHandler(sock, instanceId, tenantId, this.io, this.chatbotService)
|
||
this.msgHandlers.set(instanceId, msgHandler)
|
||
|
||
// ─── Intercepta stanzas brutas pra extrair PN real do destinatário ──
|
||
// Dois cenários distintos de mensagens do celular que chegam sem PN:
|
||
//
|
||
// Cenário A (Baileys oficial): mensagem fromMe=true com remoteJid=@lid
|
||
// O campo peer_recipient_pn na stanza bruta contém o PN do destinatário.
|
||
// Baileys não expõe esse campo em key, por isso lemos aqui.
|
||
//
|
||
// Cenário B (InfiniteAPI v7+): mensagem fromMe=true onde o InfiniteAPI
|
||
// resolve automaticamente @lid → nosso próprio JID (usando sender_pn
|
||
// da stanza, que para fromMe=true é NOSSO PN, não o do destinatário).
|
||
// O PN correto do destinatário está em recipient_pn na stanza.
|
||
// Sem essa dica, processMessage descarta a mensagem como self-chat.
|
||
//
|
||
// O listener é síncrono (mesmo EventEmitter do ws), então o mapa é
|
||
// populado antes do `messages.upsert` ser emitido de forma assíncrona.
|
||
;(sock.ws as any).on('CB:message', (node: any) => {
|
||
const msgId = node?.attrs?.id
|
||
if (!msgId) return
|
||
// peer_recipient_pn: Cenário A (@lid não resolvido, Baileys oficial)
|
||
// recipient_pn: Cenário B (InfiniteAPI resolveu @lid para nosso PN)
|
||
const hint = node?.attrs?.peer_recipient_pn ?? node?.attrs?.recipient_pn
|
||
if (hint) {
|
||
msgHandler.setPeerRecipientHint(msgId, hint)
|
||
}
|
||
})
|
||
|
||
// Presença do CONTATO (composing/recording/paused/available) → frontend nativo
|
||
// via Socket.IO (por tenant). Só chega se assinarmos (presenceSubscribe, feito na
|
||
// rota de mensagens ao abrir o chat). id = jid do chat.
|
||
;(sock.ev as any).on('presence.update', ({ id, presences }: any) => {
|
||
let state = 'available'
|
||
for (const p of Object.values(presences || {}) as any[]) {
|
||
const s = (p as any)?.lastKnownPresence
|
||
if (s === 'composing' || s === 'recording') { state = s; break }
|
||
if (s) state = s
|
||
}
|
||
this.io.to(`tenant:${tenantId}`).emit('presence:update', { instanceId, jid: id, state })
|
||
// Também via hook → ws-bridge → satélite (scoreodonto) mostra "digitando" em realtime.
|
||
hookBus.emit('ext:presence', { instanceId, tenantId, jid: id, state, timestamp: Date.now() }).catch(() => {})
|
||
})
|
||
|
||
;(sock.ev as any).on('messages.upsert', (m: any) => {
|
||
this.touchActivity(instanceId)
|
||
|
||
// Log de debug para monitoramento via página /baileys
|
||
for (const msg of m.messages ?? []) {
|
||
const text =
|
||
msg.message?.conversation ??
|
||
msg.message?.extendedTextMessage?.text ??
|
||
msg.message?.imageMessage?.caption ??
|
||
msg.message?.videoMessage?.caption ??
|
||
null
|
||
this.baileysLog(tenantId, instanceId, 'message', {
|
||
fromMe: msg.key?.fromMe,
|
||
remoteJid: msg.key?.remoteJid,
|
||
body: text,
|
||
type: Object.keys(msg.message ?? {}).find(k => k !== 'messageContextInfo') ?? 'unknown',
|
||
status: msg.status,
|
||
})
|
||
|
||
// Log de monitoramento para mensagens LID
|
||
// Ajuda a verificar se key.senderPn está presente para resolução
|
||
if (msg.key?.remoteJid?.endsWith('@lid')) {
|
||
logger.debug({
|
||
instanceId,
|
||
remoteJid: msg.key.remoteJid,
|
||
senderPn: (msg.key as any).senderPn,
|
||
pushName: msg.pushName,
|
||
}, '[LID] Mensagem @lid recebida — será resolvida pelo MessageHandler')
|
||
}
|
||
}
|
||
|
||
// A persistência do mapeamento LID→phone vive dentro do
|
||
// MessageHandler.processMessage (no upsert do Contact), porque
|
||
// updateMany aqui afetava 0 rows em contatos inexistentes — bug
|
||
// silencioso que descartava o mapeamento na primeira mensagem LID.
|
||
|
||
// Delega processamento completo para o MessageHandler
|
||
msgHandler.handle(m)
|
||
})
|
||
|
||
// ─── Status de mensagens (sent → delivered → read) ───────────────────
|
||
// O WhatsApp envia atualizações de status conforme a mensagem é entregue
|
||
// e lida. Mapeamos os códigos numéricos para strings legíveis.
|
||
;(sock.ev as any).on('messages.update', (updates: any) => {
|
||
this.touchActivity(instanceId)
|
||
for (const { key, update } of updates) {
|
||
if (!key.id || !update.status) continue
|
||
const statusMap: Record<number, string> = {
|
||
0: 'PENDING', 1: 'PENDING', 2: 'SENT', 3: 'DELIVERED', 4: 'READ', 5: 'READ',
|
||
}
|
||
const status = statusMap[update.status]
|
||
if (!status) continue
|
||
|
||
// Atualiza no banco e emite para o frontend (ticks azuis/cinza)
|
||
prisma.message.updateMany({
|
||
where: { instanceId, messageId: key.id },
|
||
data: { status: status as any },
|
||
}).then((r) => {
|
||
if (r.count === 0) return
|
||
this.io.to(`tenant:${tenantId}`).emit('message:status', {
|
||
messageId: key.id,
|
||
status,
|
||
})
|
||
hookBus.emit('ext:message.update', {
|
||
instanceId,
|
||
tenantId,
|
||
messageId: key.id,
|
||
status,
|
||
timestamp: Date.now(),
|
||
}).catch(() => {})
|
||
}).catch(() => {})
|
||
}
|
||
})
|
||
|
||
// ─── Reações recebidas ───────────────────────────────────────────────
|
||
// Listener ADITIVO: repassa a reação (emoji) da mensagem-alvo ao frontend via
|
||
// hook (ws-bridge → 'message.reaction'). text vazio = reação removida.
|
||
;(sock.ev as any).on('messages.reaction', (reactions: any) => {
|
||
this.touchActivity(instanceId)
|
||
for (const { key, reaction } of reactions) {
|
||
if (!key?.id) continue
|
||
const emoji = reaction?.text ?? ''
|
||
const sender = reaction?.key?.fromMe ? 'me' : (reaction?.key?.participant || key.participant || key.remoteJid || 'contato')
|
||
// Persiste na mensagem-alvo (para sobreviver ao reload) + emite ao vivo.
|
||
prisma.message.findFirst({ where: { instanceId, messageId: key.id }, select: { id: true, reactions: true } })
|
||
.then((m) => {
|
||
if (!m) return
|
||
const r: Record<string, string> = { ...((m.reactions as any) || {}) }
|
||
if (emoji) r[sender] = emoji; else delete r[sender]
|
||
return prisma.message.update({ where: { id: m.id }, data: { reactions: r } })
|
||
}).catch(() => {})
|
||
hookBus.emit('ext:message.reaction', {
|
||
instanceId, tenantId, messageId: key.id, emoji, sender, timestamp: Date.now(),
|
||
}).catch(() => {})
|
||
}
|
||
})
|
||
|
||
// ─── Contatos ────────────────────────────────────────────────────────
|
||
// contacts.upsert: lista completa de contatos (conexão inicial/sync)
|
||
// contacts.update: atualizações parciais (nome alterado, etc)
|
||
const contactHandler = new ContactHandler(instanceId, tenantId, sock, this.io)
|
||
this.contactHandlers.set(instanceId, contactHandler)
|
||
|
||
;(sock.ev as any).on('contacts.upsert', (contacts: any) => {
|
||
this.touchActivity(instanceId)
|
||
this.baileysLog(tenantId, instanceId, 'contacts.upsert', { count: contacts.length })
|
||
logger.info({ instanceId, count: contacts.length }, '[contacts.upsert] Recebido')
|
||
contactHandler.upsert(contacts)
|
||
})
|
||
|
||
;(sock.ev as any).on('contacts.update', (contacts: any) => {
|
||
this.touchActivity(instanceId)
|
||
this.baileysLog(tenantId, instanceId, 'contacts.update', { count: contacts.length })
|
||
contactHandler.update(contacts)
|
||
})
|
||
|
||
// ─── LID → Phone mapping (chats.phoneNumberShare) ────────────────────
|
||
// Evento emitido quando um contato envia uma mensagem de protocolo
|
||
// SHARE_PHONE_NUMBER, revelando o telefone real por trás do LID.
|
||
//
|
||
// Este é um dos mecanismos para construir o mapeamento LID → phone.
|
||
// Outros mecanismos:
|
||
// - key.senderPn nas mensagens (MessageHandler.resolveLidToPhoneJid)
|
||
// - campo `lid` nos contatos do history sync (ContactHandler.upsert)
|
||
//
|
||
// @see MessageHandler.resolveLidToPhoneJid() — consome este mapeamento
|
||
;(sock.ev as any).on('chats.phoneNumberShare', ({ lid, jid: phoneJid }: any) => {
|
||
this.touchActivity(instanceId)
|
||
logger.info({ instanceId, lid, phoneJid }, '[LID] phoneNumberShare recebido')
|
||
if (lid && phoneJid) {
|
||
// Salva o mapeamento LID no contato que já existe com o phone JID
|
||
prisma.contact.updateMany({
|
||
where: { tenantId, instanceId, jid: phoneJid },
|
||
data: { lidJid: lid },
|
||
}).catch(() => {})
|
||
|
||
// Tenta corrigir contatos que foram criados com JID = LID
|
||
// (pode acontecer se o contato chegou antes do phoneNumberShare)
|
||
prisma.contact.updateMany({
|
||
where: { tenantId, instanceId, jid: lid },
|
||
data: { jid: phoneJid, phone: phoneJid.split('@')[0], lidJid: lid },
|
||
}).catch(() => {})
|
||
}
|
||
})
|
||
|
||
// ─── Chats ───────────────────────────────────────────────────────────
|
||
// chats.upsert: novos chats (ou chats que voltaram da lista)
|
||
// chats.update: atualizações (unread count, pin, archive, etc)
|
||
;(sock.ev as any).on('chats.upsert', (chats: any) => {
|
||
this.touchActivity(instanceId)
|
||
this.baileysLog(tenantId, instanceId, 'chats.upsert', { count: chats.length })
|
||
this.syncChats(instanceId, tenantId, chats).catch(() => {})
|
||
})
|
||
|
||
;(sock.ev as any).on('chats.update', (chats: any) => {
|
||
this.touchActivity(instanceId)
|
||
this.baileysLog(tenantId, instanceId, 'chats.update', { count: chats.length })
|
||
this.handleChatsUpdate(instanceId, tenantId, chats).catch(() => {})
|
||
})
|
||
|
||
// ─── Grupos ──────────────────────────────────────────────────────────
|
||
// groups.upsert: grupo criado ou entrou em novo grupo
|
||
// groups.update: subject (nome) ou metadata alterados
|
||
// Convertemos para o formato de chat e sincronizamos via syncChats()
|
||
;(sock.ev as any).on('groups.upsert', (groups: any) => {
|
||
this.touchActivity(instanceId)
|
||
this.baileysLog(tenantId, instanceId, 'groups.upsert', { count: groups.length })
|
||
const chatsFromGroups = groups.map((g: any) => ({
|
||
id: g.id,
|
||
name: g.subject,
|
||
conversationTimestamp: g.creation,
|
||
unreadCount: 0,
|
||
pinned: false,
|
||
archived: false,
|
||
}))
|
||
this.syncChats(instanceId, tenantId, chatsFromGroups).catch(() => {})
|
||
})
|
||
|
||
;(sock.ev as any).on('groups.update', (updates: any) => {
|
||
this.touchActivity(instanceId)
|
||
this.baileysLog(tenantId, instanceId, 'groups.update', { count: updates.length })
|
||
const chatsFromUpdates = updates
|
||
.filter((g: any) => g.id && g.subject)
|
||
.map((g: any) => ({ id: g.id, name: g.subject }))
|
||
this.syncChats(instanceId, tenantId, chatsFromUpdates).catch(() => {})
|
||
})
|
||
|
||
// ─── Histórico inicial (messaging-history.set) ───────────────────────
|
||
// Evento principal de sincronização: o telefone envia chats, contatos
|
||
// e mensagens em lotes. Pode disparar múltiplas vezes.
|
||
// Quando isLatest=true, o sync terminou → dispara reconciliação de nomes.
|
||
;(sock.ev as any).on('messaging-history.set', async ({ chats: histChats, contacts: histContacts, messages: histMessages, isLatest }: any) => {
|
||
this.touchActivity(instanceId)
|
||
logger.info({
|
||
instanceId,
|
||
chats: histChats.length,
|
||
contacts: histContacts.length,
|
||
messages: histMessages.length,
|
||
isLatest,
|
||
}, '[History] messaging-history.set recebido')
|
||
|
||
this.baileysLog(tenantId, instanceId, 'messaging-history.set', {
|
||
chats: histChats.length,
|
||
contacts: histContacts.length,
|
||
messages: histMessages.length,
|
||
isLatest,
|
||
})
|
||
|
||
// Processa na ordem: contatos → chats → mensagens
|
||
// (contatos primeiro para que o vínculo Chat→Contact funcione)
|
||
if (histContacts.length > 0) {
|
||
await contactHandler.upsert(histContacts as any[]).catch(() => {})
|
||
}
|
||
if (histChats.length > 0) {
|
||
await this.syncChats(instanceId, tenantId, histChats).catch(() => {})
|
||
}
|
||
if (histMessages.length > 0) {
|
||
await msgHandler.handleHistory(histMessages).catch(() => {})
|
||
}
|
||
|
||
// Quando isLatest=true, o sync do histórico terminou completamente.
|
||
// Inicia reconciliação: busca contatos sem nome e preenche via pushName
|
||
// das mensagens recebidas (o history sync não traz pushName nos contatos).
|
||
if (isLatest) {
|
||
logger.info({ instanceId }, '[History] Sync completo (isLatest=true) — iniciando reconciliação de nomes')
|
||
contactHandler.reconcileNamesFromMessages().catch(() => {})
|
||
}
|
||
})
|
||
|
||
// ── Guard: emite DISCONNECTED se o Baileys lançar erro síncrono ─────────
|
||
// makeWASocket pode lançar em casos raros (OpenSSL, auth corrompida, etc.)
|
||
// Não há como capturar erros async do Baileys aqui além dos event listeners,
|
||
// mas garantimos que o socket foi registrado antes de prosseguir.
|
||
if (!this.sockets.has(instanceId)) {
|
||
logger.error({ instanceId }, 'Socket Baileys não foi registrado após inicialização — marcando como DISCONNECTED')
|
||
await this.updateInstanceStatus(instanceId, InstanceStatus.DISCONNECTED)
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:status', { instanceId, status: InstanceStatus.DISCONNECTED })
|
||
}
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// HANDLER DE ATUALIZAÇÃO DE CONEXÃO
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Processa eventos de mudança de estado da conexão Baileys.
|
||
*
|
||
* Estados possíveis:
|
||
* - qr: QR Code gerado → emite para o frontend escanear
|
||
* - open: conectado com sucesso → inicia sync, watchdog, etc.
|
||
* - close: desconectado → decide se reconecta ou para
|
||
* - connecting: tentando conectar → atualiza status
|
||
*
|
||
* DECISÃO DE RECONEXÃO:
|
||
* - loggedOut (401) → sessão expirada, limpa arquivos, exige novo QR
|
||
* - forbidden (403) → conta banida, para completamente
|
||
* - Qualquer outro código → reconecta com backoff exponencial
|
||
*/
|
||
private async handleConnectionUpdate(
|
||
instanceId: string,
|
||
tenantId: string,
|
||
update: Partial<ConnectionState>,
|
||
sock: WASocket
|
||
): Promise<void> {
|
||
const { connection, lastDisconnect, qr } = update
|
||
|
||
// ── QR Code emitido ───────────────────────────────────────────────────
|
||
if (qr) {
|
||
logger.info({ instanceId }, 'QR Code gerado — aguardando escaneamento')
|
||
const qrBase64 = await QRCode.toDataURL(qr)
|
||
// Salva no cache com TTL de 60s (QR expira rapidamente)
|
||
await dragonfly.set(CACHE_KEY_QR(instanceId), qrBase64, 60)
|
||
await this.updateInstanceStatus(instanceId, InstanceStatus.QR_PENDING)
|
||
// Emite para a sala da instância (página de conexão) E para o tenant (dashboard)
|
||
this.io.to(`instance:${instanceId}`).emit('qr', { instanceId, qrBase64 })
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:status', {
|
||
instanceId,
|
||
status: InstanceStatus.QR_PENDING,
|
||
qrBase64,
|
||
})
|
||
// ext: bridge — satélites externos recebem via WS /api/ext/v1/stream
|
||
hookBus.emit('ext:session.qr', { instanceId, tenantId, qrBase64, timestamp: Date.now() }).catch(() => {})
|
||
}
|
||
|
||
// ── Conectado com sucesso ─────────────────────────────────────────────
|
||
if (connection === 'open') {
|
||
logger.info({ instanceId }, 'WhatsApp conectado com sucesso')
|
||
// Reseta contadores de reconexão
|
||
this.reconnectAttempts.set(instanceId, 0)
|
||
this.clearReconnectTimer(instanceId)
|
||
// Inicia watchdog para detectar conexão morta
|
||
this.startWatchdog(instanceId)
|
||
|
||
await dragonfly.del(CACHE_KEY_QR(instanceId))
|
||
await this.updateInstanceStatus(instanceId, InstanceStatus.CONNECTED)
|
||
|
||
// Salva o número de telefone da sessão no banco
|
||
// (sock.user.id vem no formato "5511999999999:1@s.whatsapp.net")
|
||
const phoneJid = sock.user?.id
|
||
if (phoneJid) {
|
||
const phone = phoneJid.split(':')[0].split('@')[0]
|
||
await prisma.instance.update({ where: { id: instanceId }, data: { phone } })
|
||
.catch(() => {})
|
||
}
|
||
|
||
// Detecta se é conta WhatsApp Business — getBusinessProfile só retorna um
|
||
// perfil (com descrição/categoria/etc.) para contas Business. Fire-and-forget:
|
||
// não bloqueia o sync de conexão. Persiste Instance.isBusiness e avisa o front.
|
||
;(async () => {
|
||
try {
|
||
if (!phoneJid) return
|
||
const selfJid = phoneJid.split(':')[0].split('@')[0] + '@s.whatsapp.net'
|
||
const profile = await (sock as any).getBusinessProfile?.(selfJid)
|
||
const isBusiness = !!(profile && (
|
||
profile.description || profile.category || profile.email ||
|
||
profile.address || (profile.website && profile.website.length) ||
|
||
profile.business_hours
|
||
))
|
||
await prisma.instance.update({ where: { id: instanceId }, data: { isBusiness } }).catch(() => {})
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:business', { instanceId, isBusiness })
|
||
logger.info({ instanceId, isBusiness }, '[Business] detecção concluída')
|
||
} catch {
|
||
/* engine sem suporte ou conta pessoal — mantém isBusiness atual */
|
||
}
|
||
})()
|
||
|
||
// Pós-conexão: sincroniza avatar da instância e busca avatares ausentes
|
||
const handler = this.contactHandlers.get(instanceId)
|
||
if (handler) {
|
||
// Avatar da própria instância (foto de perfil do número)
|
||
handler.syncInstanceAvatar(instanceId)
|
||
.then((avatar) => {
|
||
if (avatar) {
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:avatar', { instanceId, avatar })
|
||
}
|
||
})
|
||
.catch(() => {})
|
||
|
||
// FIXADOS PRIMEIRO: puxa pin/arquivo/mute (app-state) logo e reexecuta,
|
||
// pois as app-state keys podem chegar alguns segundos após o pareamento.
|
||
setTimeout(() => handler.syncChatStates().catch(() => {}), 3000)
|
||
setTimeout(() => handler.syncChatStates().catch(() => {}), 15_000)
|
||
setTimeout(() => handler.syncChatStates().catch(() => {}), 40_000)
|
||
// Busca avatares ausentes após 5s (dá tempo para o sync inicial)
|
||
setTimeout(() => handler.syncMissingAvatars().catch(() => {}), 5000)
|
||
// Renova avatares próximos de expirar após 30s (não compete com syncMissing)
|
||
setTimeout(() => handler.syncExpiredAvatars().catch(() => {}), 30_000)
|
||
}
|
||
|
||
// Sync incremental: verifica gap de mensagens durante downtime
|
||
this.syncMissedMessages(instanceId, tenantId).catch(() => {})
|
||
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:status', {
|
||
instanceId,
|
||
status: InstanceStatus.CONNECTED,
|
||
})
|
||
hookBus.emit('ext:session.status', { instanceId, tenantId, status: 'connected', timestamp: Date.now() }).catch(() => {})
|
||
}
|
||
|
||
// ── Desconectado ──────────────────────────────────────────────────────
|
||
if (connection === 'close') {
|
||
const rawErr = lastDisconnect?.error as any
|
||
const error = (rawErr as Boom)?.output
|
||
const statusCode = error?.statusCode
|
||
const isNetworkErr = this.isTransientNetworkError(rawErr)
|
||
|
||
logger.warn({ instanceId, statusCode, isNetworkErr }, 'Conexao encerrada')
|
||
|
||
// Limpa referências da instância desconectada
|
||
this.sockets.delete(instanceId)
|
||
this.contactHandlers.delete(instanceId)
|
||
this.msgHandlers.delete(instanceId)
|
||
this.engineByInstance.delete(instanceId)
|
||
this.stopWatchdog(instanceId)
|
||
|
||
// Decide se deve reconectar baseado no código de erro
|
||
const shouldReconnect =
|
||
statusCode !== DisconnectReason.loggedOut &&
|
||
statusCode !== DisconnectReason.forbidden
|
||
|
||
if (shouldReconnect) {
|
||
const attempts = (this.reconnectAttempts.get(instanceId) ?? 0) + 1
|
||
this.reconnectAttempts.set(instanceId, attempts)
|
||
|
||
// Backoff exponencial: 3s, 6s, 12s, 24s, 48s…
|
||
// Teto menor (60s) para erros de rede transientes: DNS/ECONN se
|
||
// resolvem em segundos, não faz sentido dormir 5min. Teto geral
|
||
// (5min) fica para erros de servidor/protocolo do WhatsApp, onde
|
||
// paciência é melhor do que pressa.
|
||
const maxDelay = isNetworkErr ? RECONNECT_NETWORK_MAX_MS : RECONNECT_MAX_DELAY_MS
|
||
const delay = Math.min(
|
||
RECONNECT_BASE_DELAY_MS * Math.pow(2, attempts - 1),
|
||
maxDelay,
|
||
)
|
||
|
||
const scheduledAt = Date.now() + delay
|
||
const nextAt = new Date(scheduledAt).toISOString()
|
||
logger.info({ instanceId, attempts, delayMs: delay, nextAt, isNetworkErr },
|
||
'[Reconexao] Agendando reconexao com backoff exponencial')
|
||
|
||
await this.updateInstanceStatus(instanceId, InstanceStatus.CONNECTING)
|
||
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:status', {
|
||
instanceId,
|
||
status: InstanceStatus.CONNECTING,
|
||
reconnectIn: delay,
|
||
attempt: attempts,
|
||
})
|
||
|
||
this.scheduleReconnect(instanceId, tenantId, delay)
|
||
} else {
|
||
// loggedOut → limpa sessão e exige novo QR
|
||
if (statusCode === DisconnectReason.loggedOut) {
|
||
logger.warn({ instanceId }, 'Sessao desconectada pelo usuario — limpando sessao')
|
||
await this.deleteSessionFiles(instanceId)
|
||
}
|
||
// forbidden → conta banida, apenas atualiza status
|
||
await this.updateInstanceStatus(instanceId, InstanceStatus.DISCONNECTED)
|
||
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:status', {
|
||
instanceId,
|
||
status: InstanceStatus.DISCONNECTED,
|
||
reason: (DisconnectReason as unknown as Record<number, string>)[statusCode] ?? statusCode,
|
||
})
|
||
hookBus.emit('ext:session.status', { instanceId, tenantId, status: 'disconnected', timestamp: Date.now() }).catch(() => {})
|
||
}
|
||
}
|
||
|
||
// ── Conectando ────────────────────────────────────────────────────────
|
||
if (connection === 'connecting') {
|
||
await this.updateInstanceStatus(instanceId, InstanceStatus.CONNECTING)
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:status', {
|
||
instanceId,
|
||
status: InstanceStatus.CONNECTING,
|
||
})
|
||
}
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// SYNC INCREMENTAL — Mensagens perdidas durante downtime
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Verifica e notifica sobre gap de mensagens durante downtime.
|
||
*
|
||
* Consulta a última mensagem no banco e calcula quanto tempo o servidor
|
||
* ficou offline. O Baileys com syncFullHistory=true já solicita as
|
||
* mensagens pendentes automaticamente — este método apenas:
|
||
* 1. Loga o gap para visibilidade (monitoramento)
|
||
* 2. Emite 'instance:syncing' para o frontend mostrar indicador de sync
|
||
*/
|
||
private async syncMissedMessages(instanceId: string, tenantId: string): Promise<void> {
|
||
try {
|
||
const lastMsg = await prisma.message.findFirst({
|
||
where: { instanceId },
|
||
orderBy: { timestamp: 'desc' },
|
||
select: { timestamp: true, messageId: true },
|
||
})
|
||
|
||
if (!lastMsg) {
|
||
logger.info({ instanceId }, '[Sync] Nenhuma mensagem no DB — history sync completo via Baileys')
|
||
return
|
||
}
|
||
|
||
const gap = Date.now() - lastMsg.timestamp.getTime()
|
||
const gapMinutes = Math.round(gap / 60_000)
|
||
|
||
logger.info({ instanceId, lastMsgAt: lastMsg.timestamp, gapMinutes },
|
||
'[Sync] Reconectado — gap de mensagens detectado')
|
||
|
||
this.baileysLog(tenantId, instanceId, 'sync:gap', {
|
||
lastMsgAt: lastMsg.timestamp.toISOString(),
|
||
gapMinutes,
|
||
})
|
||
|
||
// Notifica o frontend que está sincronizando
|
||
this.io.to(`tenant:${tenantId}`).emit('instance:syncing', {
|
||
instanceId,
|
||
gapMinutes,
|
||
since: lastMsg.timestamp.toISOString(),
|
||
})
|
||
} catch (err) {
|
||
logger.error({ err, instanceId }, '[Sync] Erro ao verificar gap de mensagens')
|
||
}
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// WATCHDOG — Detecta conexão morta silenciosamente
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Registra timestamp da última atividade para o watchdog.
|
||
* Chamado por TODOS os event listeners do Baileys.
|
||
*/
|
||
private touchActivity(instanceId: string): void {
|
||
this.lastActivity.set(instanceId, Date.now())
|
||
}
|
||
|
||
/**
|
||
* Inicia o watchdog para uma instância.
|
||
*
|
||
* Fluxo conservador em 3 etapas — escolhido pra minimizar reconexões:
|
||
* 1. Se inativo > TIMEOUT (10 min): envia um presence update como
|
||
* ping leve e marca timestamp. Se o socket estiver vivo, Baileys
|
||
* responde e touchActivity é chamado no próximo evento.
|
||
* 2. Se continua inativo GRACE (+30s) após o ping: confirma que a
|
||
* conexão está de fato morta. Procede pro kill.
|
||
* 3. Antes de matar: verifica cool-down (15 min desde último kill)
|
||
* pra não entrar em loop quando a rede está oscilando. Se estiver
|
||
* no cool-down, só loga e espera o próximo ciclo.
|
||
*
|
||
* Esta sequência dá ~10-11 min de folga antes do primeiro kill, e no
|
||
* mínimo 15 min entre kills consecutivos — 7-8× mais conservador que
|
||
* a versão anterior (2 min sem verificação de estado).
|
||
*/
|
||
private startWatchdog(instanceId: string): void {
|
||
this.stopWatchdog(instanceId)
|
||
|
||
const timer = setInterval(async () => {
|
||
const now = Date.now()
|
||
const lastTs = this.lastActivity.get(instanceId) ?? 0
|
||
const elapsed = now - lastTs
|
||
const tenantId = this.tenantByInstance.get(instanceId)
|
||
|
||
if (elapsed <= WATCHDOG_TIMEOUT_MS) {
|
||
// Conexão com atividade recente — reseta estado de ping/cool-down
|
||
this.lastWatchdogPing.delete(instanceId)
|
||
return
|
||
}
|
||
|
||
const sock = this.sockets.get(instanceId)
|
||
|
||
// Caso 1: socket sumiu sem connection.close — reconecta imediatamente
|
||
if (!sock) {
|
||
if (tenantId) {
|
||
logger.warn({ instanceId },
|
||
'[Watchdog] Socket ausente — reagendando conexao')
|
||
this.stopWatchdog(instanceId)
|
||
this.connect(instanceId, tenantId).catch(() => {})
|
||
}
|
||
return
|
||
}
|
||
|
||
const pingedAt = this.lastWatchdogPing.get(instanceId) ?? 0
|
||
const sincePing = now - pingedAt
|
||
|
||
// Etapa 1: envia ping leve (presence update) se ainda não pingou
|
||
// neste ciclo de inatividade. Não conta como atividade — só estímulo.
|
||
if (pingedAt === 0 || sincePing > WATCHDOG_TIMEOUT_MS) {
|
||
try {
|
||
await sock.sendPresenceUpdate('available')
|
||
this.lastWatchdogPing.set(instanceId, now)
|
||
logger.debug({ instanceId, elapsedMs: elapsed },
|
||
'[Watchdog] Ping leve enviado — aguardando resposta')
|
||
} catch (err) {
|
||
// Ping falhou → socket já está morto, pula pra etapa de kill
|
||
logger.warn({ instanceId, err },
|
||
'[Watchdog] Ping falhou — socket morto')
|
||
this.forceKillSocket(instanceId, sock, tenantId, elapsed)
|
||
}
|
||
return
|
||
}
|
||
|
||
// Etapa 2: ping já foi enviado, mas ainda sem atividade após GRACE.
|
||
// InfiniteAPI pode ter ACK mais lento — usa tolerância maior.
|
||
const engine = this.engineByInstance.get(instanceId) ?? 'infinite'
|
||
const graceMs = engine === 'infinite' ? WATCHDOG_PING_GRACE_INFIN_MS : WATCHDOG_PING_GRACE_MS
|
||
if (sincePing > graceMs) {
|
||
this.forceKillSocket(instanceId, sock, tenantId, elapsed)
|
||
}
|
||
}, WATCHDOG_INTERVAL_MS)
|
||
|
||
this.watchdogTimers.set(instanceId, timer)
|
||
}
|
||
|
||
/**
|
||
* Força o fechamento do socket, respeitando cool-down para evitar loop.
|
||
* Só mata se não houve outro kill nos últimos WATCHDOG_COOLDOWN_MS.
|
||
*/
|
||
private forceKillSocket(
|
||
instanceId: string,
|
||
sock: WASocket,
|
||
tenantId: string | undefined,
|
||
elapsedMs: number,
|
||
): void {
|
||
const now = Date.now()
|
||
const lastKill = this.lastWatchdogKill.get(instanceId) ?? 0
|
||
const sinceKill = now - lastKill
|
||
|
||
if (sinceKill < WATCHDOG_COOLDOWN_MS) {
|
||
logger.warn(
|
||
{ instanceId, elapsedMs, sinceLastKillMs: sinceKill },
|
||
'[Watchdog] Em cool-down — adiando kill para proteger numero de ban',
|
||
)
|
||
return
|
||
}
|
||
|
||
logger.warn({ instanceId, elapsedMs },
|
||
'[Watchdog] Conexao morta confirmada — forcando reconexao')
|
||
|
||
this.lastWatchdogKill.set(instanceId, now)
|
||
this.lastWatchdogPing.delete(instanceId)
|
||
|
||
try { sock.end(new Error('Watchdog timeout')) } catch { /* ignore */ }
|
||
|
||
// Se o sock.end não disparar connection.close (raro), reconecta manualmente
|
||
if (!this.sockets.has(instanceId) && tenantId) {
|
||
this.stopWatchdog(instanceId)
|
||
this.connect(instanceId, tenantId).catch(() => {})
|
||
}
|
||
}
|
||
|
||
/** Para o watchdog de uma instância */
|
||
private stopWatchdog(instanceId: string): void {
|
||
const timer = this.watchdogTimers.get(instanceId)
|
||
if (timer) {
|
||
clearInterval(timer)
|
||
this.watchdogTimers.delete(instanceId)
|
||
}
|
||
// Limpa estado de ping pendente — novo ciclo começa limpo.
|
||
// NÃO limpa lastWatchdogKill aqui: o cool-down deve sobreviver a
|
||
// stop/start para bloquear reconexões em cascata.
|
||
this.lastWatchdogPing.delete(instanceId)
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// TIMER DE RECONEXÃO
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/** Cancela timer de reconexão pendente (evita reconexão duplicada) */
|
||
private clearReconnectTimer(instanceId: string): void {
|
||
const timer = this.reconnectTimers.get(instanceId)
|
||
if (timer) {
|
||
clearTimeout(timer)
|
||
this.reconnectTimers.delete(instanceId)
|
||
}
|
||
this.reconnectScheduledAt.delete(instanceId)
|
||
}
|
||
|
||
/**
|
||
* Agenda uma reconexão após `delay` ms.
|
||
*
|
||
* Centraliza o setTimeout + tracking do scheduledAt (usado pelo network
|
||
* probe pra saber se vale adiantar). Cancela qualquer timer anterior.
|
||
*/
|
||
private scheduleReconnect(instanceId: string, tenantId: string, delay: number): void {
|
||
this.clearReconnectTimer(instanceId)
|
||
const scheduledAt = Date.now() + delay
|
||
this.reconnectScheduledAt.set(instanceId, scheduledAt)
|
||
const timer = setTimeout(() => {
|
||
this.reconnectScheduledAt.delete(instanceId)
|
||
this.connect(instanceId, tenantId)
|
||
}, delay)
|
||
this.reconnectTimers.set(instanceId, timer)
|
||
}
|
||
|
||
/**
|
||
* Classifica um erro de conexão como "rede transiente".
|
||
*
|
||
* Erros de DNS/TCP costumam resolver em segundos. Erros de status do
|
||
* WhatsApp (loggedOut, conflict, restartRequired) são categoria diferente
|
||
* e devem usar o backoff longo tradicional.
|
||
*/
|
||
private isTransientNetworkError(err: any): boolean {
|
||
if (!err) return false
|
||
const code = err.code || err.cause?.code || err.errno
|
||
if (code && typeof code === 'string') {
|
||
const transient = [
|
||
'EAI_AGAIN', // DNS retry
|
||
'ENOTFOUND', // DNS não resolveu
|
||
'ECONNRESET',
|
||
'ECONNREFUSED',
|
||
'ETIMEDOUT',
|
||
'EHOSTUNREACH',
|
||
'ENETUNREACH',
|
||
'ENETDOWN',
|
||
'EPIPE',
|
||
]
|
||
if (transient.includes(code)) return true
|
||
}
|
||
// Fallback: mensagem contém nome do host — é quase certo que é DNS/WS
|
||
const msg = (err.message || '').toString()
|
||
if (/getaddrinfo|web\.whatsapp\.com|WebSocket Error/i.test(msg)) return true
|
||
return false
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// NETWORK PROBE — Acorda reconexões quando a rede volta
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Inicia o network probe global.
|
||
*
|
||
* A cada NETWORK_PROBE_INTERVAL_MS, testa se o DNS resolve
|
||
* `web.whatsapp.com`. Se resolver E houver alguma instância com retry
|
||
* agendado para daqui a mais de PROBE_ADVANCE_THRESHOLD_MS, cancela o
|
||
* timer dessa instância e dispara o retry imediatamente.
|
||
*
|
||
* Por quê DNS em vez de HTTP? Zero bytes para o WhatsApp. DNS falha com
|
||
* EAI_AGAIN quando a rede está caída, e resolve rápido quando volta —
|
||
* é um sinal confiável de disponibilidade de rede sem tocar no WA.
|
||
*/
|
||
private startNetworkProbe(): void {
|
||
if (this.networkProbeTimer) return
|
||
this.networkProbeTimer = setInterval(async () => {
|
||
// Curto-circuito: se não há retries pendentes, nem faz o DNS.
|
||
if (this.reconnectScheduledAt.size === 0) return
|
||
|
||
try {
|
||
await dns.lookup(PROBE_DNS_HOST)
|
||
} catch {
|
||
return // Rede ainda caída — não adianta nada, espera o tick seguinte.
|
||
}
|
||
|
||
// DNS OK → adianta qualquer retry que ainda falte > threshold
|
||
const now = Date.now()
|
||
for (const [instanceId, scheduledAt] of this.reconnectScheduledAt) {
|
||
const remaining = scheduledAt - now
|
||
if (remaining <= PROBE_ADVANCE_THRESHOLD_MS) continue
|
||
const tenantId = this.tenantByInstance.get(instanceId)
|
||
if (!tenantId) continue
|
||
logger.info({ instanceId, remainingMs: remaining },
|
||
'[NetworkProbe] Rede detectada — adiantando retry pendente')
|
||
this.clearReconnectTimer(instanceId)
|
||
// Dispara sem await — probe roda em interval e não deve bloquear
|
||
this.connect(instanceId, tenantId).catch((err) =>
|
||
logger.error({ err, instanceId }, '[NetworkProbe] Falha ao adiantar retry'),
|
||
)
|
||
}
|
||
}, NETWORK_PROBE_INTERVAL_MS)
|
||
// Não segura o processo vivo só por causa do probe
|
||
this.networkProbeTimer.unref?.()
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// DESCONEXÃO MANUAL
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Desconecta uma instância manualmente (logout).
|
||
*
|
||
* Diferente de uma desconexão por erro — aqui o usuário está
|
||
* intencionalmente desconectando. Limpa todos os timers e
|
||
* faz logout no WhatsApp (invalidando a sessão).
|
||
*/
|
||
async disconnect(instanceId: string): Promise<void> {
|
||
const sock = this.sockets.get(instanceId)
|
||
if (!sock) return
|
||
this.clearReconnectTimer(instanceId)
|
||
this.stopWatchdog(instanceId)
|
||
this.reconnectAttempts.delete(instanceId)
|
||
// Logout manual zera o estado de cool-down: reconexão futura é um
|
||
// novo ciclo e não deve herdar o bloqueio do watchdog.
|
||
this.lastWatchdogKill.delete(instanceId)
|
||
this.lastActivity.delete(instanceId)
|
||
await sock.logout()
|
||
this.sockets.delete(instanceId)
|
||
this.contactHandlers.delete(instanceId)
|
||
this.msgHandlers.delete(instanceId)
|
||
this.engineByInstance.delete(instanceId)
|
||
logger.info({ instanceId }, 'Instancia desconectada manualmente')
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// ACESSORES PÚBLICOS
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/** Retorna o socket Baileys ativo para uma instância (usado pelas rotas de envio) */
|
||
getSocket(instanceId: string): WASocket | undefined {
|
||
return this.sockets.get(instanceId)
|
||
}
|
||
|
||
/** Retorna o Socket.IO server (usado pelas rotas para emitir eventos) */
|
||
getIO(): SocketIOServer {
|
||
return this.io
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// HELPERS INTERNOS
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Atualiza o status da instância no banco E no cache.
|
||
* Dupla escrita para que o frontend possa consultar status tanto
|
||
* via API (banco) quanto via polling rápido (cache DragonflyDB).
|
||
*/
|
||
private async updateInstanceStatus(
|
||
instanceId: string,
|
||
status: InstanceStatus
|
||
): Promise<void> {
|
||
await Promise.all([
|
||
prisma.instance.update({ where: { id: instanceId }, data: { status } })
|
||
.catch((err: any) => { if (err?.code !== 'P2025') throw err }),
|
||
dragonfly.set(CACHE_KEY_STATUS(instanceId), status),
|
||
])
|
||
}
|
||
|
||
/** Remove arquivos de sessão do disco (após loggedOut) */
|
||
public async deleteSessionFiles(instanceId: string): Promise<void> {
|
||
const sessionDir = path.resolve(env.BAILEYS_SESSIONS_PATH, instanceId)
|
||
await fs.rm(sessionDir, { recursive: true, force: true })
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// SINCRONIZAÇÃO DE CHATS
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Sincroniza lista de chats no banco.
|
||
*
|
||
* Chamado por: chats.upsert, chats.update, groups.upsert, groups.update,
|
||
* messaging-history.set
|
||
*
|
||
* Para cada chat:
|
||
* - Cria ou atualiza no banco com JID, nome (grupo), timestamps
|
||
* - Vincula ao Contact existente (se houver)
|
||
* - Sincroniza estado de pin/archive/unread do WhatsApp
|
||
*
|
||
* Otimização: busca todos os contatos existentes em uma única query
|
||
* e usa Map para lookup O(1) por JID.
|
||
*/
|
||
private async syncChats(instanceId: string, tenantId: string, chats: any[]): Promise<void> {
|
||
if (chats.length === 0) return
|
||
|
||
// Busca todos os contatos existentes para os JIDs (1 query)
|
||
const jids = chats.map((c) => c.id).filter(Boolean)
|
||
const existingContacts = await prisma.contact.findMany({
|
||
where: { tenantId, instanceId, jid: { in: jids } },
|
||
select: { id: true, jid: true },
|
||
})
|
||
const contactIdByJid = new Map(existingContacts.map((c) => [c.jid, c.id]))
|
||
|
||
// Pré-carrega mapeamento LID → phone JID para chats @lid.
|
||
// ContactHandler.upsert já rodou antes (messaging-history.set processa
|
||
// contacts → chats → messages nessa ordem), então lidJid já está populado.
|
||
const lidJids = jids.filter((j) => j.endsWith('@lid'))
|
||
const lidToPhoneJid = new Map<string, string>() // @lid → @s.whatsapp.net
|
||
const lidToContactId = new Map<string, string>() // @lid → contactId
|
||
if (lidJids.length > 0) {
|
||
const resolvedByLid = await prisma.contact.findMany({
|
||
where: {
|
||
tenantId,
|
||
instanceId,
|
||
lidJid: { in: lidJids },
|
||
NOT: { jid: { endsWith: '@lid' } }, // só aceita se resolveu para phone JID
|
||
},
|
||
select: { id: true, jid: true, lidJid: true },
|
||
})
|
||
for (const rc of resolvedByLid) {
|
||
if (rc.lidJid) {
|
||
lidToPhoneJid.set(rc.lidJid, rc.jid)
|
||
lidToContactId.set(rc.lidJid, rc.id)
|
||
}
|
||
}
|
||
}
|
||
|
||
for (const chat of chats) {
|
||
const rawJid = chat.id
|
||
if (!rawJid) continue
|
||
|
||
// Descarta JIDs de sistema: broadcasts, newsletters e conta PSA (0@s.whatsapp.net)
|
||
if (rawJid.includes('@broadcast') || rawJid.endsWith('@newsletter') || rawJid.startsWith('0@')) continue
|
||
|
||
// ── Resolução LID → phone JID ─────────────────────────────────────
|
||
// Chats @lid são artefatos do protocolo WhatsApp multi-device:
|
||
// o JID real do contato é @s.whatsapp.net.
|
||
// Fontes de resolução (ordem de prioridade):
|
||
// 1. Mapeamento no banco (ContactHandler.upsert populou via lidJid)
|
||
// 2. Campo pnJid do objeto de chat — enviado pelo InfiniteAPI v7+ no
|
||
// messaging-history.set direto do proto.IConversation.pnJid
|
||
let jid = rawJid
|
||
if (rawJid.endsWith('@lid')) {
|
||
const phoneJid = lidToPhoneJid.get(rawJid)
|
||
?? (chat as any).pnJid as string | undefined
|
||
if (phoneJid) {
|
||
jid = phoneJid
|
||
// Garante que o contactId mapeado seja usado abaixo
|
||
const resolvedContactId = lidToContactId.get(rawJid) ?? contactIdByJid.get(jid)
|
||
if (resolvedContactId) contactIdByJid.set(jid, resolvedContactId)
|
||
// Persiste o mapeamento no banco caso tenha vindo só do pnJid (fallback)
|
||
if (!lidToPhoneJid.has(rawJid)) {
|
||
prisma.contact.updateMany({
|
||
where: { tenantId, instanceId, jid: phoneJid },
|
||
data: { lidJid: rawJid },
|
||
}).catch(() => {})
|
||
}
|
||
} else {
|
||
// Sem resolução disponível: chat será criado via handleHistory quando
|
||
// a mensagem chegar com senderPn resolvível.
|
||
logger.debug({ rawJid, instanceId }, '[syncChats] @lid sem resolução — descartando')
|
||
continue
|
||
}
|
||
}
|
||
|
||
try {
|
||
// Converte timestamp Unix (segundos) para Date.
|
||
// Fallback: lastMessageRecvTimestamp (InfiniteAPI define a partir das mensagens do sync)
|
||
// quando conversationTimestamp é 0 ou null (ex: chats antigos fora da janela de sync).
|
||
const rawTs = Number(chat.conversationTimestamp || 0)
|
||
|| Number((chat as any).lastMessageRecvTimestamp || 0)
|
||
const lastAt = rawTs > 0 ? new Date(rawTs * 1000) : undefined
|
||
|
||
const isGroup = jid.endsWith('@g.us')
|
||
// Nome do chat: apenas grupos têm nome (subject) — individuais usam Contact.name
|
||
const groupName = isGroup ? (chat.name ?? null) : null
|
||
const contactId = contactIdByJid.get(jid) ?? null
|
||
|
||
await prisma.chat.upsert({
|
||
where: { tenantId_instanceId_jid: { tenantId, instanceId, jid } },
|
||
create: {
|
||
tenantId,
|
||
instanceId,
|
||
jid,
|
||
name: groupName,
|
||
unreadCount: chat.unreadCount ?? 0,
|
||
lastMessageAt: lastAt ?? null,
|
||
isPinned: !!chat.pinned,
|
||
isArchived: chat.archived ?? false,
|
||
...(contactId && { contactId }),
|
||
},
|
||
update: {
|
||
...(groupName && { name: groupName }),
|
||
...(contactId && { contactId }),
|
||
// Só aplica campos que vieram explicitamente no evento.
|
||
// chats.upsert pode ser parcial (nova mensagem, metadata) — se
|
||
// pinned/archived/unreadCount estiverem ausentes, não sobrescreve
|
||
// o valor salvo no banco.
|
||
...(chat.unreadCount !== undefined && { unreadCount: chat.unreadCount }),
|
||
...(lastAt && { lastMessageAt: lastAt }),
|
||
...(chat.pinned !== undefined && { isPinned: !!chat.pinned }),
|
||
...(chat.archived !== undefined && { isArchived: !!chat.archived }),
|
||
},
|
||
})
|
||
} catch (err) {
|
||
logger.error({ err, jid }, '[WhatsApp] Erro ao sincronizar chat')
|
||
}
|
||
}
|
||
// Invalida cache da listagem de chats — refletir alterações imediatamente na inbox
|
||
invalidateChatListCache(tenantId, instanceId).catch(() => {})
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// HANDLER DE ATUALIZAÇÕES PARCIAIS DE CHAT (chats.update)
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Processa atualizações parciais de chats (pin, archive, unread).
|
||
*
|
||
* Diferente de syncChats (usado para bulk/history), este método:
|
||
* 1. Só atualiza campos presentes na atualização (sem sobrescrever com defaults)
|
||
* 2. Emite chat:upsert via Socket.IO para atualizar o frontend em tempo real
|
||
*
|
||
* O evento chats.update do Baileys traz objetos parciais — apenas os campos
|
||
* que mudaram. Por isso não podemos usar ?? 0 para unreadCount, pois isso
|
||
* sobrescreveria o valor correto com zero.
|
||
*/
|
||
private async handleChatsUpdate(instanceId: string, tenantId: string, updates: any[]): Promise<void> {
|
||
for (const update of updates) {
|
||
const jid = update.id
|
||
if (!jid || jid.endsWith('@lid')) continue
|
||
|
||
try {
|
||
// Monta patch com apenas os campos presentes na atualização
|
||
const patch: Record<string, any> = {}
|
||
if (update.pinned !== undefined) patch.isPinned = !!update.pinned
|
||
if (update.archived !== undefined) patch.isArchived = !!update.archived
|
||
if (update.unreadCount !== undefined) patch.unreadCount = update.unreadCount
|
||
|
||
if (Object.keys(patch).length === 0) continue
|
||
|
||
const chat = await prisma.chat.update({
|
||
where: { tenantId_instanceId_jid: { tenantId, instanceId, jid } },
|
||
data: patch,
|
||
include: {
|
||
contact: {
|
||
select: {
|
||
name: true, verifiedName: true, notify: true,
|
||
phone: true, avatarUrl: true,
|
||
scoreReputacao: true, flagRestricao: true,
|
||
},
|
||
},
|
||
},
|
||
}).catch(() => null)
|
||
|
||
if (!chat) continue
|
||
|
||
// Emite para o frontend atualizar em tempo real
|
||
const c = chat.contact
|
||
const isGroup = jid.endsWith('@g.us')
|
||
const displayName = isGroup
|
||
? (chat.name ?? null)
|
||
: (c?.name ?? c?.verifiedName ?? c?.notify ?? null)
|
||
|
||
this.io.to(`tenant:${tenantId}`).emit('chat:upsert', {
|
||
id: chat.id,
|
||
tenantId: chat.tenantId,
|
||
instanceId: chat.instanceId,
|
||
jid: chat.jid,
|
||
name: chat.name ?? null,
|
||
unreadCount: chat.unreadCount,
|
||
lastMessageAt: chat.lastMessageAt,
|
||
isPinned: chat.isPinned,
|
||
isArchived: chat.isArchived,
|
||
botPaused: chat.botPaused,
|
||
contact: c
|
||
? {
|
||
name: displayName,
|
||
phone: c.phone,
|
||
avatarUrl: c.avatarUrl,
|
||
scoreReputacao: c.scoreReputacao,
|
||
flagRestricao: c.flagRestricao,
|
||
}
|
||
: isGroup
|
||
? { name: displayName, phone: null, avatarUrl: null, scoreReputacao: 0, flagRestricao: false }
|
||
: null,
|
||
lastMessage: null,
|
||
})
|
||
} catch (err) {
|
||
logger.error({ err, jid }, '[WhatsApp] Erro ao aplicar chats.update')
|
||
}
|
||
}
|
||
invalidateChatListCache(tenantId, instanceId).catch(() => {})
|
||
}
|
||
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
// DEBUG LOG (página /baileys no frontend)
|
||
// ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
/**
|
||
* Emite evento de debug para a página /baileys do frontend.
|
||
*
|
||
* Permite monitorar em tempo real todos os eventos Baileys
|
||
* sem precisar acessar os logs do servidor.
|
||
*/
|
||
private baileysLog(tenantId: string, instanceId: string, event: string, data: unknown): void {
|
||
this.io.to(`tenant:${tenantId}`).emit('baileys:log', {
|
||
timestamp: new Date().toISOString(),
|
||
sessionId: instanceId,
|
||
event,
|
||
data,
|
||
})
|
||
}
|
||
}
|