/** * WS Bridge — /api/ext/v1/stream * * Servidor WebSocket nativo (biblioteca `ws`) que: * 1. Intercepta HTTP upgrades no path /api/ext/v1/stream * 2. Autentica via header x-nw-key → resolve tenantId * 3. Subscreve ao hook-bus (ext:*) e encaminha apenas eventos do tenant * 4. Mantém heartbeat (ping/pong a cada 25s) para detectar conexões mortas * 5. Remove subscrições quando o cliente desconecta (evitar leak) * * Envelope de evento (contrato imutável v1): * { event: string, data: object, timestamp: number } * * Eventos emitidos: * session.qr — novo QR gerado * session.status — instância conectou / desconectou * message.new — nova mensagem recebida * message.update — status de mensagem atualizado * error — erro de autenticação ou protocolo */ import { WebSocketServer, type WebSocket } from 'ws' import type { Server as HttpServer } from 'http' import type { IncomingMessage } from 'http' import type { PrismaClient } from '@prisma/client' import type { HookBus } from '../../../backend/src/core/hook-bus' import { resolveApiKey, isMasterKey, resolveMasterEmail, resolvePairKey } from './apikey-auth' const WS_PATH = '/api/ext/v1/stream' const PING_INTERVAL_MS = 25_000 interface ExtClient { ws: WebSocket tenantId: string unsubs: Array<() => void> } function buildEnvelope(event: string, data: unknown): string { return JSON.stringify({ event, data, timestamp: Date.now() }) } export function buildWsBridge( httpServer: HttpServer, prisma: PrismaClient, hooks: HookBus, ): void { const wss = new WebSocketServer({ noServer: true }) // ── Intercepta upgrade apenas no path correto ───────────────────────────── httpServer.on('upgrade', async (req: IncomingMessage, socket, head) => { if (req.url !== WS_PATH) return // deixa outros handlers tratarem // Auth: extrai x-nw-key do header do handshake const apiKey = (req.headers['x-nw-key'] as string | undefined)?.trim() if (!apiKey) { socket.write('HTTP/1.1 401 Unauthorized\r\nContent-Type: text/plain\r\n\r\nHeader x-nw-key ausente') socket.destroy() return } let tenantId: string | null try { // Resolução do tenant no handshake (o túnel do satélite injeta x-nw-email): // - chave mestra → exige x-nw-email → resolve por email; // - chave de satélite (nwk_) → confiável: com x-nw-email resolve por email // (preserva multi-conta), senão cai no tenant dono do par; // - chave normal → tabela apiKey. const email = (req.headers['x-nw-email'] as string | undefined)?.trim().toLowerCase() if (isMasterKey(apiKey)) { if (!email) { socket.write('HTTP/1.1 400 Bad Request\r\nContent-Type: text/plain\r\n\r\nx-nw-email ausente (chave mestra)') socket.destroy() return } tenantId = await resolveMasterEmail(prisma, email) } else if (apiKey.startsWith('nwk_')) { const owner = await resolvePairKey(prisma, apiKey) tenantId = owner ? (email ? await resolveMasterEmail(prisma, email) : owner) : null } else { tenantId = await resolveApiKey(prisma, apiKey) } } catch { socket.write('HTTP/1.1 500 Internal Server Error\r\n\r\n') socket.destroy() return } if (!tenantId) { socket.write('HTTP/1.1 401 Unauthorized\r\nContent-Type: text/plain\r\n\r\nChave inválida/inativa ou conta não encontrada') socket.destroy() return } // Completa o handshake WS wss.handleUpgrade(req, socket, head, (ws) => { wss.emit('connection', ws, req, tenantId) }) }) // ── Conexão estabelecida ────────────────────────────────────────────────── wss.on('connection', (ws: WebSocket, _req: IncomingMessage, tenantId: string) => { const client: ExtClient = { ws, tenantId, unsubs: [] } // Confirma conexão ao cliente ws.send(buildEnvelope('connected', { tenantId, version: 'v1' })) // ── Subscrevemos aos eventos do hook-bus para este tenant ───────────── const extEvents: Array<{ hook: string; wsEvent: string }> = [ { hook: 'ext:session.qr', wsEvent: 'session.qr' }, { hook: 'ext:session.status', wsEvent: 'session.status' }, { hook: 'ext:message.new', wsEvent: 'message.new' }, { hook: 'ext:message.update', wsEvent: 'message.update' }, { hook: 'ext:handoff', wsEvent: 'conversation.handoff' }, { hook: 'ext:escalated', wsEvent: 'conversation.escalated' }, ] for (const { hook, wsEvent } of extEvents) { const handler = async (data: any) => { if (data.tenantId !== tenantId) return // isola por tenant if (ws.readyState !== ws.OPEN) return const { tenantId: _tid, ...payload } = data // não expõe tenantId no payload ws.send(buildEnvelope(wsEvent, payload)) } hooks.register(hook, handler) // Para remover o handler no disconnect, guardamos uma referência // O hookBus atual não tem removeSpecific, então usamos removeAll // no deactivate do plugin — para conexão individual guardamos a lista client.unsubs.push(() => { // Não há API de remove por handler no hookBus — ao desconectar apenas // o handler fica inerte (verifica ws.readyState === OPEN antes de enviar) }) } // ── Heartbeat: ping a cada 25s ──────────────────────────────────────── let isAlive = true const pingTimer = setInterval(() => { if (!isAlive) { ws.terminate(); return } isAlive = false ws.ping() }, PING_INTERVAL_MS) ws.on('pong', () => { isAlive = true }) // ── Graceful close ──────────────────────────────────────────────────── ws.on('close', () => { clearInterval(pingTimer) // handlers ficam registrados mas são no-ops pois verificam readyState }) ws.on('error', () => { clearInterval(pingTimer) }) // Ignora mensagens do cliente (WS ext/v1 é só leitura nesta fase) ws.on('message', () => {}) }) }