/** * 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 } 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 { 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 expirada') 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', () => {}) }) }