feat(realtime): monitoramento em tempo real via WebSocket multi-tenant

Fecha agente.md secao 54-55 (infraestrutura) e 161 (WebSocket multi-tenant).
Entrega o pipeline de push em tempo real completo — o consumo visual
("Monitoramento -> Filas/Ramais") fica pra fase Frontend.

Requisito central da secao 161 ("nao transmitir tudo e filtrar so no
browser"): RealtimeGateway tem um unico ponto de emissao,
broadcastToTenant(), sempre server.to(`tenant:<id>`), nunca broadcast
global. Cada socket entra na room do proprio tenant no handshake, nunca
escolhe a room.

Autenticacao na conexao (handshake.auth.token, nao Authorization header):
valida o JWT (mesmo verifyAccessToken do JwtAuthGuard), exige tenantId no
token e a permission monitoring.view (ja existia desde RBAC, sem
consumidor ate agora) — mesmo principio de nunca confiar em tenant_id do
client, so do JWT ja emitido por /auth/select-tenant.

Origem dos eventos: canal Redis unico b2bcall:events (o mesmo desde Event
Socket). Dois produtores: b2bcall-fs-events (eventos do FreeSWITCH,
resolvendo tenantId por fan-out quando nao ha channel variable, ver
tenant-resolve.ts) e apps/api (mudancas no nosso Agent.state via
agents-me.controller, tenantId direto do JWT, sem fan-out).

Bug real achado e corrigido ao construir esta fase: nenhum evento CUSTOM do
ESL (sofia::register, sofia::gateway_state, callcenter::info) jamais
chegava em b2bcall-fs-events nesta sessao inteira. Causa: event_json(...)
mandava "CUSTOM" como ultimo token do comando `event json`, sem subclass
depois — mod_event_socket exige os subclasses logo depois do token CUSTOM
no mesmo comando pra serem entregues. Corrigido separando PLAIN_EVENTS
(viram listener .on()) de CUSTOM_SUBCLASSES (so compoem o comando de
assinatura). Resolve as lacunas ja documentadas em docs/TRUNKS.md e
docs/AGENTS.md. De quebra, corrigido um bug de nome de campo
(CC-Agent-Status, que nao existe -> CC-Agent-State) e um segundo bug real
em trunk-sync.ts (rescan nunca descarregava gateway removido -> agora roda
`killgw` antes do rescan).

Novos tipos normalizados a partir de callcenter::info, com nomes de campo
confirmados contra uma fila real: AGENT_OFFERED_CALL, AGENT_BRIDGE_FAILED,
QUEUE_MEMBER_COUNT (chamadas esperando, secao 54), QUEUE_MEMBER_LEFT (com
cause/cancelReason e timestamps — base pra Service Level/Abandon Rate
quando CDR existir).

Verificado ponta a ponta com um client socket.io real: login/pause/resume/
logout emitindo AGENT_STATE_CHANGED; chamada de teste numa fila com agente
logado emitindo QUEUE_MEMBER_COUNT/LEFT, AGENT_OFFERED_CALL,
AGENT_BRIDGE_FAILED, AGENT_STATUS_CHANGED (CC-Agent-State correto); token
ausente/invalido desconectado na hora, sem vazar nenhum evento.

Achado sistemico durante o teste (documentado, nao corrigido nesta fase):
@@unique combinado com soft delete, sem excluir deletedAt, em
Agent/Extension/Trunk/Queue/PauseReason — nao da pra reusar numero/nome/
codigo depois de apagar. Precisa de indice unico parcial em cada um, fora
do escopo desta fase.

typecheck do workspace inteiro limpo. ~144MB de memoria total nos
containers (fs-events 44MB, fs-config 45MB, freeswitch 26MB, postgres
21MB, redis 8MB).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01X1HxY46WGU4G1zmVDNKcWw

apps/api/package.json

@@ -19,13 +19,16 @@
     "@nestjs/common": "^12.0.1",
     "@nestjs/core": "^12.0.1",
     "@nestjs/platform-fastify": "^12.0.1",
+    "@nestjs/platform-socket.io": "12.0.1",
+    "@nestjs/websockets": "12.0.1",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.15.1",
     "fast-xml-parser": "5.11.1",
     "fastify": "5.12.1",
     "ioredis": "^6.0.0",
     "reflect-metadata": "^0.2.2",
-    "rxjs": "^7.8.2"
+    "rxjs": "^7.8.2",
+    "socket.io": "4.8.3"
   },
   "devDependencies": {
     "@types/node": "^22.0.0",

apps/api/src/agents/agents-me.controller.ts

@@ -13,6 +13,7 @@ import { JwtAuthGuard } from "../common/guards/jwt-auth.guard";
 import { CurrentUser } from "../common/decorators/current-user.decorator";
 import { PauseDto } from "./dto/pause.dto";
 import { notifyAgentChanged, notifyTierChanged } from "./agent-sync.helper";
+import { publishAgentStateChanged } from "../realtime/realtime-publish.helper";
 
 async function findMyAgent(tx: Prisma.TransactionClient, tenantId: string, userId: string) {
   const agent = await tx.agent.findFirst({
@@ -69,6 +70,7 @@ export class AgentsMeController {
     for (const tier of agent.tiers) {
       await notifyTierChanged(tenantId, tier.queueId, agent.id, "upsert", tier.level, tier.position);
     }
+    await publishAgentStateChanged(tenantId, agent.id, "AVAILABLE");
 
     return { state: "AVAILABLE" };
   }
@@ -99,6 +101,7 @@ export class AgentsMeController {
     await recordAuditEvent(prisma, { action: "AGENT_LOGOUT", tenantId, userId: user.sub, entityType: "agent", entityId: agent.id });
 
     await notifyAgentChanged(tenantId, agent.id, "upsert");
+    await publishAgentStateChanged(tenantId, agent.id, "OFFLINE");
 
     return { state: "OFFLINE" };
   }
@@ -141,6 +144,7 @@ export class AgentsMeController {
     });
 
     await notifyAgentChanged(tenantId, agent.id, "upsert");
+    await publishAgentStateChanged(tenantId, agent.id, "PAUSED");
 
     return { state: "PAUSED" };
   }
@@ -167,6 +171,7 @@ export class AgentsMeController {
     await recordAuditEvent(prisma, { action: "AGENT_RESUME", tenantId, userId: user.sub, entityType: "agent", entityId: agent.id });
 
     await notifyAgentChanged(tenantId, agent.id, "upsert");
+    await publishAgentStateChanged(tenantId, agent.id, "AVAILABLE");
 
     return { state: "AVAILABLE" };
   }

apps/api/src/app.module.ts

@@ -7,6 +7,7 @@ import { DialplanModule } from "./dialplan/dialplan.module";
 import { QueuesModule } from "./queues/queues.module";
 import { AgentsModule } from "./agents/agents.module";
 import { PauseReasonsModule } from "./pause-reasons/pause-reasons.module";
+import { RealtimeModule } from "./realtime/realtime.module";
 
 @Module({
   imports: [
@@ -18,6 +19,7 @@ import { PauseReasonsModule } from "./pause-reasons/pause-reasons.module";
     QueuesModule,
     AgentsModule,
     PauseReasonsModule,
+    RealtimeModule,
   ],
 })
 export class AppModule {}

apps/api/src/main.ts

@@ -1,6 +1,7 @@
 import "reflect-metadata";
 import { NestFactory } from "@nestjs/core";
 import { FastifyAdapter, type NestFastifyApplication } from "@nestjs/platform-fastify";
+import { IoAdapter } from "@nestjs/platform-socket.io";
 import { ValidationPipe } from "@nestjs/common";
 import helmet from "@fastify/helmet";
 import cors from "@fastify/cors";
@@ -42,6 +43,11 @@ async function bootstrap() {
 
   app.useGlobalFilters(new DomainExceptionFilter());
 
+  // Monitoramento em tempo real (agente.md secao 54-55, 161) — socket.io
+  // sobre o mesmo servidor HTTP do Fastify, path proprio pra nao colidir
+  // com as rotas REST.
+  app.useWebSocketAdapter(new IoAdapter(app));
+
   const port = Number(process.env.API_PORT ?? 3000);
   await app.listen(port, "127.0.0.1");
   console.log(`b2bcall-api ouvindo em http://127.0.0.1:${port}`);

apps/api/src/realtime/realtime-publish.helper.ts

@@ -0,0 +1,25 @@
+import type { NormalizedEvent } from "@b2bcall/telephony";
+import { getRedisClient } from "../common/redis";
+
+const EVENTS_CHANNEL = "b2bcall:events";
+
+/**
+ * Publica no mesmo canal que b2bcall-fs-events usa pros eventos vindos do
+ * FreeSWITCH — o RealtimeRedisBridge (apps/api) não distingue a origem, só
+ * reencaminha pro tenant certo via WebSocket (docs/REALTIME.md). Diferente
+ * dos eventos vindos do ESL, aqui o tenantId já é conhecido de cara (vem do
+ * JWT da requisição HTTP) — não precisa do fan-out de tenant-resolve.ts.
+ */
+export async function publishAgentStateChanged(
+  tenantId: string,
+  agentId: string,
+  state: string,
+): Promise<void> {
+  const event: NormalizedEvent = {
+    type: "AGENT_STATE_CHANGED",
+    occurredAt: new Date().toISOString(),
+    tenantId,
+    data: { agentId, state },
+  };
+  await getRedisClient().publish(EVENTS_CHANNEL, JSON.stringify(event));
+}

apps/api/src/realtime/realtime-redis-bridge.service.ts

@@ -0,0 +1,52 @@
+import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from "@nestjs/common";
+import Redis from "ioredis";
+import type { NormalizedEvent } from "@b2bcall/telephony";
+import { RealtimeGateway } from "./realtime.gateway";
+
+const EVENTS_CHANNEL = "b2bcall:events";
+
+/**
+ * Pub/sub exige uma conexão Redis dedicada em modo subscriber (não pode
+ * compartilhar com o client usado pra comandos normais) — mesmo padrão já
+ * usado em fs-config/fs-events.
+ */
+@Injectable()
+export class RealtimeRedisBridge implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(RealtimeRedisBridge.name);
+  private subscriber?: Redis;
+
+  constructor(private readonly gateway: RealtimeGateway) {}
+
+  async onModuleInit(): Promise<void> {
+    const url = process.env.REDIS_URL;
+    if (!url) {
+      throw new Error("REDIS_URL nao configurado");
+    }
+
+    this.subscriber = new Redis(url);
+    this.subscriber.on("error", (err) => this.logger.error(`erro na conexao Redis (subscriber): ${String(err)}`));
+    await this.subscriber.subscribe(EVENTS_CHANNEL);
+
+    this.subscriber.on("message", (_channel, raw) => {
+      let event: NormalizedEvent;
+      try {
+        event = JSON.parse(raw);
+      } catch (err) {
+        this.logger.error(`evento invalido no canal ${EVENTS_CHANNEL}: ${String(err)}`);
+        return;
+      }
+      if (!event.tenantId) {
+        // Sem tenant resolvido (ex.: chamada sintética sem agente/fila real
+        // por trás) — nada pra emitir, nenhuma room saberia receber.
+        return;
+      }
+      this.gateway.broadcastToTenant(event.tenantId, event);
+    });
+
+    this.logger.log("bridge Redis -> WebSocket conectado");
+  }
+
+  async onModuleDestroy(): Promise<void> {
+    await this.subscriber?.quit();
+  }
+}

apps/api/src/realtime/realtime.gateway.ts

@@ -0,0 +1,64 @@
+import { Injectable, Logger } from "@nestjs/common";
+import { OnGatewayConnection, OnGatewayDisconnect, WebSocketGateway, WebSocketServer } from "@nestjs/websockets";
+import type { Server, Socket } from "socket.io";
+import { verifyAccessToken, userHasPermission } from "@b2bcall/auth";
+import type { NormalizedEvent } from "@b2bcall/telephony";
+
+function tenantRoom(tenantId: string): string {
+  return `tenant:${tenantId}`;
+}
+
+/**
+ * Monitoramento em tempo real (agente.md secao 54-55, 161: "eventos
+ * WebSocket devem ser tenant-scoped no servidor, nunca transmitir tudo e
+ * filtrar só no browser"). `broadcastToTenant` é o único ponto de emissão —
+ * sempre `server.to(room)`, nunca `server.emit()` global.
+ */
+@Injectable()
+@WebSocketGateway({
+  cors: {
+    origin: process.env.CORS_ORIGIN ? process.env.CORS_ORIGIN.split(",") : false,
+    credentials: true,
+  },
+  path: "/realtime",
+})
+export class RealtimeGateway implements OnGatewayConnection, OnGatewayDisconnect {
+  private readonly logger = new Logger(RealtimeGateway.name);
+
+  @WebSocketServer()
+  server!: Server;
+
+  async handleConnection(client: Socket): Promise<void> {
+    const token = client.handshake.auth?.token as string | undefined;
+    if (!token) {
+      client.disconnect(true);
+      return;
+    }
+
+    try {
+      const claims = await verifyAccessToken(token);
+      if (!claims.tenantId) {
+        client.disconnect(true);
+        return;
+      }
+      const allowed = await userHasPermission(claims.sub, "monitoring.view", claims.tenantId);
+      if (!allowed) {
+        client.disconnect(true);
+        return;
+      }
+
+      await client.join(tenantRoom(claims.tenantId));
+      this.logger.log(`cliente conectado ao monitoramento (tenant ${claims.tenantId})`);
+    } catch {
+      client.disconnect(true);
+    }
+  }
+
+  handleDisconnect(): void {
+    // socket.io já limpa as rooms sozinho na desconexão — nada a fazer aqui.
+  }
+
+  broadcastToTenant(tenantId: string, event: NormalizedEvent): void {
+    this.server.to(tenantRoom(tenantId)).emit("event", event);
+  }
+}

apps/api/src/realtime/realtime.module.ts

@@ -0,0 +1,8 @@
+import { Module } from "@nestjs/common";
+import { RealtimeGateway } from "./realtime.gateway";
+import { RealtimeRedisBridge } from "./realtime-redis-bridge.service";
+
+@Module({
+  providers: [RealtimeGateway, RealtimeRedisBridge],
+})
+export class RealtimeModule {}

apps/freeswitch-events/src/main.ts

@@ -3,6 +3,7 @@ import type { FreeSwitchEventData } from "esl";
 import { FreeSwitchTelephonyProvider, normalizeEslEvent } from "@b2bcall/telephony";
 import { createLogger } from "@b2bcall/shared";
 import { updateTrunkStatusFromGatewayEvent } from "./trunk-status";
+import { resolveTenantIdForAgent, resolveTenantIdForQueue } from "./tenant-resolve";
 
 const logger = createLogger("b2bcall-fs-events");
 
@@ -97,7 +98,27 @@ async function main() {
     logger.error("erro no client ESL", { error: String(err) });
   });
 
-  function handleEvent(eventName: string, raw: FreeSwitchEventData) {
+  // callcenter::info/sofia::gateway_state não carregam b2bcall_tenant_id
+  // (só existe como channel variable a partir do Predictive Engine) — pra
+  // esses tipos, o tenant é resolvido pelo id do agente/fila embutido no
+  // nome FreeSWITCH (tenant-resolve.ts). Sem isso, o WebSocket multi-tenant
+  // (docs/REALTIME.md) não teria como saber pra qual tenant emitir.
+  async function resolveTenantId(normalized: ReturnType<typeof normalizeEslEvent>): Promise<string | undefined> {
+    if (!normalized || normalized.tenantId) return normalized?.tenantId;
+    switch (normalized.type) {
+      case "AGENT_STATUS_CHANGED":
+      case "AGENT_OFFERED_CALL":
+      case "AGENT_BRIDGE_FAILED":
+        return resolveTenantIdForAgent(normalized.data.agent as string | undefined);
+      case "QUEUE_MEMBER_COUNT":
+      case "QUEUE_MEMBER_LEFT":
+        return resolveTenantIdForQueue(normalized.data.queue as string | undefined);
+      default:
+        return undefined;
+    }
+  }
+
+  async function handleEvent(eventName: string, raw: FreeSwitchEventData) {
     // Para eventos JSON (event_json), os campos reais do evento FreeSWITCH
     // (Event-Name, Unique-ID, Event-Subclass, variable_*, ...) vem em
     // `raw.body`; `raw.headers` são só os headers do protocolo ESL.
@@ -107,6 +128,12 @@ async function main() {
       return;
     }
 
+    try {
+      normalized.tenantId = await resolveTenantId(normalized);
+    } catch (err) {
+      logger.error("falha ao resolver tenant do evento", { error: String(err), type: normalized.type });
+    }
+
     redis.publish(REDIS_CHANNEL, JSON.stringify(normalized)).catch((err) => {
       logger.error("falha ao publicar evento normalizado no Redis", { error: String(err) });
     });

apps/freeswitch-events/src/tenant-resolve.ts

@@ -0,0 +1,49 @@
+import { getPrismaClient, withTenantContext, type Prisma } from "@b2bcall/database";
+
+/**
+ * Eventos de callcenter/gateway não carregam tenant_id (só existe como
+ * channel variable a partir do Predictive Engine, ver normalize-event.ts) —
+ * mas o id do FreeSWITCH (`<uuid>@dominio`) É o id primário da nossa linha
+ * (Agent.id/Queue.id), então dá pra achar o tenant dono procurando em cada
+ * tenant ativo (mesmo padrão de fan-out de trunk-status.ts). Cacheado por id
+ * — um agente/fila nunca muda de tenant, então uma entrada no cache nunca
+ * fica desatualizada (só inútil, se o registro for apagado depois).
+ */
+const cache = new Map<string, string>();
+
+function extractId(fsName: string | undefined): string | undefined {
+  if (!fsName) return undefined;
+  return fsName.split("@")[0];
+}
+
+async function resolveViaFanOut(
+  cacheKey: string,
+  lookup: (tx: Prisma.TransactionClient, tenantId: string) => Promise<unknown>,
+): Promise<string | undefined> {
+  const cached = cache.get(cacheKey);
+  if (cached) return cached;
+
+  const prisma = getPrismaClient();
+  const tenants = await prisma.tenant.findMany({ where: { status: "ACTIVE" }, select: { id: true } });
+
+  for (const tenant of tenants) {
+    const found = await withTenantContext(prisma, tenant.id, (tx) => lookup(tx, tenant.id));
+    if (found) {
+      cache.set(cacheKey, tenant.id);
+      return tenant.id;
+    }
+  }
+  return undefined;
+}
+
+export async function resolveTenantIdForAgent(agentFsName: string | undefined): Promise<string | undefined> {
+  const id = extractId(agentFsName);
+  if (!id) return undefined;
+  return resolveViaFanOut(`agent:${id}`, (tx) => tx.agent.findUnique({ where: { id } }));
+}
+
+export async function resolveTenantIdForQueue(queueFsName: string | undefined): Promise<string | undefined> {
+  const id = extractId(queueFsName);
+  if (!id) return undefined;
+  return resolveViaFanOut(`queue:${id}`, (tx) => tx.queue.findUnique({ where: { id } }));
+}