From cb6d343b2e62728309e0284b1fc1dd40e4e69dfc Mon Sep 17 00:00:00 2001 From: Matheus Date: Fri, 28 Aug 2026 13:31:31 -0300 Subject: [PATCH] feat(dialer): CPS Limiter + Predictive Dialer Engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fecha agente.md secao 72-86 (motor preditivo) e 77-79 (CPS distribuido, reserva de leads, lock de campanha). Uma campanha RUNNING agora origina chamadas sozinha, respeitando capacidade de agentes, CPS hierarquico e taxa de abandono — sem intervencao manual. Deliberadamente fora do escopo (agente.md secao 72: "nao e' so' `for lead -> originate`"): mod_avmd (opcional), callbacks agendados, disposicoes de agente — ficam pra fase CDR. ## Novo servico apps/predictive-dialer Mesmo padrao arquitetural de fs-events/fs-config: Node standalone em Docker, ESL propria, tick a cada 2s sobre tenants ativos x campanhas RUNNING/WAITING_SCHEDULE. - Lock de campanha (dialer:campaign:{id}, secao 79): TTL/ownership/ renewal/safe-release via Lua compare-and-delete. - CPS distribuido (secao 77, 62): token bucket janela 1s, hierarquia GLOBAL/TENANT/TRUNK/CAMPAIGN numa unica chamada Lua atomica — nivel esgotado bloqueia todos SEM incremento parcial dos que passariam. - Reserva atomica de leads (secao 78): FOR UPDATE SKIP LOCKED dentro da mesma transacao withTenantContext. - CallAttempt/CampaignStats (schema novo): state machine da chamada (secao 82) + EWMA (secao 75) de answer_probability/average_answer_delay/ average_talk_time/abandon_rate por campanha. - Capacidade em tempo real + pacing (secao 73-76, 84-85): conta agentes por estado via Tier->Agent.state, previsao de liberacao (horizonte unico de 15s, simplificacao documentada dos 4 buckets da especificacao), controle de abandono reduz pacing progressivamente, nunca origina sem capacidade prevista. ## Modo simulacao (secao 185-186) DIALER_SIMULATION=true (default, ja estava no .env desde o inicio da sessao) sorteia ANSWER/BUSY/NO_ANSWER/FAILED em software, sem PSTN real. So' quando ANSWERED e' que uma chamada sintetica (null/dummy, sem PSTN) entra na fila real via mod_callcenter de verdade — escolha deliberada pra maximizar codigo real exercitado em vez de simular tudo em memoria. Os identificadores da secao 81 (b2bcall_tenant_id/call_id/attempt_id/ campaign_id/lead_id) vao como channel variables nessa perna, entregando tenantId real no WebSocket sem fan-out. Real Outbound Safety (secao 186): as duas flags checadas no boot, nunca ativadas automaticamente — caminho PSTN real implementado mas nunca exercitado (sem trunk/operadora real neste laboratorio). ## Dois bugs reais achados e corrigidos testando esta fase - Perna sintetica (null/dummy) nao tem midia do outro lado — nunca desligava sozinha depois de bridgear com um agente. Corrigido com hangup agendado via uuid_kill no talk_time simulado. - Corrida entre queue:sync e tier:sync (dois canais Redis independentes, sem ordem garantida): atribuir tier logo depois de criar a fila podia rodar tier add antes do queue reload terminar ("-ERR Queue not found!", erro real, diferente do ja conhecido "already exist"). Corrigido com retry curto (ate 3 tentativas) em agent-sync.ts::addTierWithRetry. ## GET /campaigns/:id/stats Secao 227.7 "visualizar pacing" — CampaignStats + agentes por estado + calls em andamento, sem esperar a fase Frontend. Verificado ponta a ponta: campanha RUNNING originando 3 tentativas por tick, outcomes simulados corretos com retry agendado (BUSY 15min/ NO_ANSWER 60min/FAILED 30min), uma tentativa ANSWERED completando o ciclo real inteiro (fila -> agente -> bridge -> hangup -> EWMA atualizada), stop nao derruba chamada ativa (secao 66), calls_answered=3 confirmado no `queue list` do FreeSWITCH. CPS limiter e lock de campanha testados isoladamente (hierarquia sem incremento parcial, ownership nunca roubado). typecheck do workspace inteiro limpo. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01X1HxY46WGU4G1zmVDNKcWw --- TODO.md | 63 ++++- .../api/src/campaigns/campaigns.controller.ts | 51 ++++ apps/freeswitch-config/src/agent-sync.ts | 36 ++- apps/predictive-dialer/Dockerfile | 26 ++ apps/predictive-dialer/package.json | 23 ++ apps/predictive-dialer/src/call-attempt.ts | 137 ++++++++++ apps/predictive-dialer/src/event-listener.ts | 140 ++++++++++ apps/predictive-dialer/src/ewma.ts | 12 + .../predictive-dialer/src/lead-reservation.ts | 51 ++++ apps/predictive-dialer/src/main.ts | 134 ++++++++++ apps/predictive-dialer/src/originate.ts | 75 ++++++ apps/predictive-dialer/src/pacing.ts | 155 +++++++++++ .../src/queued-attempts-registry.ts | 44 ++++ .../predictive-dialer/src/redis-primitives.ts | 98 +++++++ apps/predictive-dialer/src/retry-rules.ts | 16 ++ apps/predictive-dialer/src/schedule.ts | 43 +++ apps/predictive-dialer/src/simulation.ts | 53 ++++ apps/predictive-dialer/src/tick.ts | 244 ++++++++++++++++++ apps/predictive-dialer/tsconfig.json | 9 + docker-compose.yml | 22 ++ docs/CAMPAIGNS.md | 9 +- docs/PREDICTIVE_DIALER.md | 197 ++++++++++++++ .../migration.sql | 60 +++++ .../migration.sql | 14 + .../migration.sql | 10 + packages/database/prisma/schema.prisma | 122 ++++++++- packages/telephony/src/normalize-event.ts | 1 + packages/telephony/src/types.ts | 6 +- pnpm-lock.yaml | 28 ++ 29 files changed, 1851 insertions(+), 28 deletions(-) create mode 100644 apps/predictive-dialer/Dockerfile create mode 100644 apps/predictive-dialer/package.json create mode 100644 apps/predictive-dialer/src/call-attempt.ts create mode 100644 apps/predictive-dialer/src/event-listener.ts create mode 100644 apps/predictive-dialer/src/ewma.ts create mode 100644 apps/predictive-dialer/src/lead-reservation.ts create mode 100644 apps/predictive-dialer/src/main.ts create mode 100644 apps/predictive-dialer/src/originate.ts create mode 100644 apps/predictive-dialer/src/pacing.ts create mode 100644 apps/predictive-dialer/src/queued-attempts-registry.ts create mode 100644 apps/predictive-dialer/src/redis-primitives.ts create mode 100644 apps/predictive-dialer/src/retry-rules.ts create mode 100644 apps/predictive-dialer/src/schedule.ts create mode 100644 apps/predictive-dialer/src/simulation.ts create mode 100644 apps/predictive-dialer/src/tick.ts create mode 100644 apps/predictive-dialer/tsconfig.json create mode 100644 docs/PREDICTIVE_DIALER.md create mode 100644 packages/database/prisma/migrations/20260828155954_call_attempts_campaign_stats/migration.sql create mode 100644 packages/database/prisma/migrations/20260828160116_add_tenant_to_campaign_stats/migration.sql create mode 100644 packages/database/prisma/migrations/20260828160500_call_attempts_campaign_stats_rls/migration.sql diff --git a/TODO.md b/TODO.md index 8ad3b9d..c8dc297 100644 --- a/TODO.md +++ b/TODO.md @@ -317,12 +317,69 @@ - [ ] `PredictiveDialerEngine` (secao 72-86) — dados em tempo real, EWMA, CPS distribuído, reserva atômica de lead, lock de campanha, originate via bgapi, state machine da chamada, controle de abandono, - retry — fase própria, não iniciada + retry — implementado na PHASE 16 - [ ] Wizard visual de importação (upload de arquivo) — fase Frontend - [ ] Relatório de campanha (secao 160) — depende de CDR -## PHASE 16+ — ver `agente.md` seções 72 em diante (Predictive Dialer Engine, -CDR, Recordings, AI, Billing, Frontend, Reports, Security, Tests) +## PHASE 16 — CPS Limiter / Predictive Dialer Engine (agente.md secao 72-86, 77-79) +- [x] Novo serviço `apps/predictive-dialer` (mesmo padrão de fs-events/ + fs-config: Node standalone em Docker, ESL própria, tick a cada 2s + sobre tenants ativos x campanhas RUNNING/WAITING_SCHEDULE) +- [x] Lock de campanha (`dialer:campaign:{id}`, secao 79): TTL, ownership + token, renewal, safe release (Lua compare-and-delete) — testado + isolado, outro dono nunca rouba nem renova o lock de quem já tem +- [x] CPS distribuído (secao 77, 62): token bucket janela de 1s (Lua + atômico), hierarquia GLOBAL/TENANT/TRUNK/CAMPAIGN numa única chamada + — testado isolado, nível esgotado bloqueia todos SEM incremento + parcial dos que passariam +- [x] Reserva atômica de leads (secao 78): `FOR UPDATE SKIP LOCKED` raw + SQL dentro da mesma transação `withTenantContext` +- [x] `CallAttempt`/`CampaignStats` (schema novo): state machine da + chamada (secao 82) + EWMA (secao 75, alpha=0.25) de + answer_probability/average_answer_delay/average_talk_time/ + abandon_rate, persistida por campanha +- [x] Capacidade em tempo real + pacing (secao 73-76, 84-85): conta + agentes por estado via Tier->Agent.state, previsão de liberação + simplificada (horizonte único de 15s em vez dos 4 buckets da + especificação), controle de abandono reduz pacing progressivamente, + nunca origina sem capacidade prevista +- [x] Modo simulação (secao 185): DIALER_SIMULATION=true (default) sorteia + ANSWER/BUSY/NO_ANSWER/FAILED em software, sem PSTN real. Só quando + ANSWERED é que uma chamada sintética (null/dummy, sem PSTN) entra na + fila real via mod_callcenter de verdade — maximiza código real + exercitado em vez de simular tudo em memória +- [x] Real Outbound Safety (secao 186): as duas flags + (DIALER_SIMULATION=false + ALLOW_REAL_OUTBOUND_CALLS=true) checadas + no boot, nunca ativado automaticamente — caminho PSTN real + implementado mas nunca exercitado (sem trunk real neste laboratório) +- [x] **Bug real, achado no teste desta fase**: perna sintética (null/ + dummy) não tem mídia do outro lado — nunca desliga sozinha depois de + bridgear com um agente. Corrigido com hangup agendado via + `uuid_kill` no talk_time simulado. +- [x] **Bug real, achado no teste desta fase**: corrida entre `queue:sync` + e `tier:sync` (dois canais Redis independentes, sem ordem garantida) + — atribuir tier logo depois de criar a fila podia rodar `tier add` + antes do `queue reload` terminar ("-ERR Queue not found!", diferente + do já conhecido "already exist"). Corrigido com retry curto em + `agent-sync.ts::addTierWithRetry`. +- [x] `GET /campaigns/:id/stats` (secao 227.7 "visualizar pacing"): + CampaignStats + contagem de agentes por estado + calls em andamento +- [x] Testado ponta a ponta: campanha RUNNING originando 3 tentativas por + tick, outcomes simulados corretos com retry agendado, uma chamada + simulada-ANSWERED completando o ciclo real inteiro (queue -> agente + -> bridge -> hangup -> EWMA atualizada), stop não derruba chamada + ativa (secao 66), calls_answered=3 confirmado no `queue list` do + FreeSWITCH ao final +- [ ] Previsão de 4 buckets (5/10/15/20s, secao 74) — simplificado pra um + único horizonte de 15s +- [ ] mod_avmd (secao 87), callbacks agendados (secao 88), disposições de + agente (secao 89) — fora do escopo, ficam pra fase CDR +- [ ] CPS a nível de trunk — só aplica quando o caminho PSTN real rodar +- [ ] Relatório de campanha (secao 160), TME/TMA/Service Level/Abandon + Rate agregados — dependem de CDR + +## PHASE 17+ — ver `agente.md` seções 87 em diante (CDR, Recordings, AI, +Billing, Frontend, Reports, Security, Tests) --- diff --git a/apps/api/src/campaigns/campaigns.controller.ts b/apps/api/src/campaigns/campaigns.controller.ts index 4f6a73d..2b9398f 100644 --- a/apps/api/src/campaigns/campaigns.controller.ts +++ b/apps/api/src/campaigns/campaigns.controller.ts @@ -132,6 +132,57 @@ export class CampaignsController { return campaign; } + /** Pacing ao vivo (agente.md secao 227.7 "visualizar pacing"): estatísticas + * EWMA persistidas pelo PredictiveDialerEngine a cada tick + uma contagem + * de agentes por estado calculada na hora (mesma fonte de verdade que a + * fase Realtime Monitoring já usa, Agent.state). */ + @RequirePermission("campaigns.view") + @Get(":id/stats") + async stats(@CurrentUser() user: AccessTokenClaims, @Param("id") id: string) { + const prisma = getPrismaClient(); + const tenantId = user.tenantId!; + + const campaign = await withTenantContext(prisma, tenantId, (tx) => + tx.campaign.findFirst({ where: { id, tenantId, deletedAt: null } }), + ); + if (!campaign) throw new NotFoundException(); + + const [stats, agents, callsInFlight] = await withTenantContext(prisma, tenantId, (tx) => + Promise.all([ + tx.campaignStats.findUnique({ where: { campaignId: id } }), + tx.agent.findMany({ + where: { tenantId, enabled: true, deletedAt: null, tiers: { some: { queueId: campaign.queueId } } }, + select: { state: true }, + }), + tx.callAttempt.count({ + where: { + tenantId, + campaignId: id, + status: { in: ["CREATED", "RESERVED", "ORIGINATING", "ORIGINATED", "RINGING", "ANSWERED", "QUEUEING"] }, + }, + }), + ]), + ); + + const agentsByState: Record = {}; + for (const agent of agents) { + agentsByState[agent.state] = (agentsByState[agent.state] ?? 0) + 1; + } + + return { + status: campaign.status, + stats: stats ?? { + answerProbability: null, + averageAnswerDelay: null, + averageTalkTime: null, + abandonRate: null, + pacingFactor: campaign.pacingInitial, + }, + agentsByState, + callsInFlight, + }; + } + @RequirePermission("campaigns.start") @Post(":id/start") async start(@CurrentUser() user: AccessTokenClaims, @Param("id") id: string) { diff --git a/apps/freeswitch-config/src/agent-sync.ts b/apps/freeswitch-config/src/agent-sync.ts index 0239c4d..de8e0ba 100644 --- a/apps/freeswitch-config/src/agent-sync.ts +++ b/apps/freeswitch-config/src/agent-sync.ts @@ -127,9 +127,39 @@ export async function syncTier(msg: TierSyncMessage): Promise { return; } - await providerInstance - .addAgentToQueue(queueName, agentName, msg.level ?? 1, msg.position ?? 1) - .catch(() => undefined); // "already exist" e' esperado/ok + await addTierWithRetry(providerInstance, queueName, agentName, msg.level ?? 1, msg.position ?? 1); logger.info("tier sincronizado", { queueName, agentName }); } + +/** + * Achado real: `queues:sync` (que faz a fila existir de verdade no + * FreeSWITCH via reloadxml+queue reload) e `tiers:sync` são dois canais + * Redis independentes, sem ordem garantida entre si — se um tier for + * atribuído logo depois de criar a fila (ex.: script de setup rápido, + * fase Predictive Engine), o `tier add` pode rodar antes do `queue reload` + * terminar, e falha com "-ERR Queue not found!" (erro real, diferente de + * "already exist", que é o único que antes era silenciosamente ignorado + * aqui). Retry curto com backoff cobre essa corrida sem precisar + * coordenar os dois canais. + */ +async function addTierWithRetry( + providerInstance: FreeSwitchTelephonyProvider, + queueName: string, + agentName: string, + level: number, + position: number, + attempt = 1, +): Promise { + try { + await providerInstance.addAgentToQueue(queueName, agentName, level, position); + } catch (err) { + const body = (err as { res?: { body?: string } }).res?.body ?? ""; + if (body.includes("already exist")) return; // esperado/ok num resync + if (body.includes("Queue not found") && attempt < 4) { + await new Promise((resolve) => setTimeout(resolve, 500 * attempt)); + return addTierWithRetry(providerInstance, queueName, agentName, level, position, attempt + 1); + } + logger.error("falha ao adicionar tier apos retries", { queueName, agentName, attempt, error: body || String(err) }); + } +} diff --git a/apps/predictive-dialer/Dockerfile b/apps/predictive-dialer/Dockerfile new file mode 100644 index 0000000..b5348cb --- /dev/null +++ b/apps/predictive-dialer/Dockerfile @@ -0,0 +1,26 @@ +# syntax=docker/dockerfile:1.7 +# +# Mesmo padrão de apps/freeswitch-events/Dockerfile: roda via `tsx` direto +# (sem etapa de `tsc build`), build a partir da raiz do monorepo. +FROM node:22-slim + +RUN corepack enable && corepack prepare pnpm@11.24.0 --activate + +WORKDIR /repo + +COPY pnpm-workspace.yaml package.json pnpm-lock.yaml tsconfig.base.json ./ +COPY packages/types packages/types +COPY packages/shared packages/shared +COPY packages/telephony packages/telephony +COPY packages/database packages/database +COPY apps/predictive-dialer apps/predictive-dialer + +RUN pnpm install --frozen-lockfile --filter @b2bcall/predictive-dialer... + +# `prisma generate` só precisa do schema, não de uma conexão real. +ENV DATABASE_URL="postgresql://placeholder:placeholder@localhost:5432/placeholder" +RUN pnpm --filter @b2bcall/database exec prisma generate + +WORKDIR /repo/apps/predictive-dialer + +CMD ["pnpm", "exec", "tsx", "src/main.ts"] diff --git a/apps/predictive-dialer/package.json b/apps/predictive-dialer/package.json new file mode 100644 index 0000000..c2a4b8c --- /dev/null +++ b/apps/predictive-dialer/package.json @@ -0,0 +1,23 @@ +{ + "name": "@b2bcall/predictive-dialer", + "version": "0.0.1", + "private": true, + "scripts": { + "dev": "tsx watch src/main.ts", + "build": "tsc -p tsconfig.json", + "start": "node dist/main.js", + "typecheck": "tsc --noEmit" + }, + "dependencies": { + "@b2bcall/database": "workspace:*", + "@b2bcall/shared": "workspace:*", + "@b2bcall/telephony": "workspace:*", + "esl": "11.2.1", + "ioredis": "^6.0.0" + }, + "devDependencies": { + "@types/node": "^22.0.0", + "tsx": "^4.23.12", + "typescript": "^5.7.0" + } +} diff --git a/apps/predictive-dialer/src/call-attempt.ts b/apps/predictive-dialer/src/call-attempt.ts new file mode 100644 index 0000000..d1e4517 --- /dev/null +++ b/apps/predictive-dialer/src/call-attempt.ts @@ -0,0 +1,137 @@ +import { randomUUID } from "node:crypto"; +import type { CallAttempt, CallAttemptStatus, Prisma } from "@b2bcall/database"; +import { computeNextAttemptAt } from "./retry-rules"; +import { ewmaUpdate } from "./ewma"; + +export interface CreateAttemptParams { + tenantId: string; + campaignId: string; + leadId: string; + simulated: boolean; +} + +export async function createCallAttempt( + tx: Prisma.TransactionClient, + params: CreateAttemptParams, +): Promise { + return tx.callAttempt.create({ + data: { + tenantId: params.tenantId, + campaignId: params.campaignId, + leadId: params.leadId, + simulated: params.simulated, + status: "RESERVED", + }, + }); +} + +export async function setAttemptStatus( + tx: Prisma.TransactionClient, + attemptId: string, + status: CallAttemptStatus, + extra: Partial> = {}, +): Promise { + await tx.callAttempt.update({ where: { id: attemptId }, data: { status, ...extra } }); +} + +export function newOriginationUuid(): string { + return randomUUID(); +} + +export type TerminalOutcome = "COMPLETED" | "BUSY" | "NO_ANSWER" | "FAILED" | "ABANDONED"; + +export interface CompleteAttemptParams { + attemptId: string; + tenantId: string; + campaignId: string; + leadId: string; + outcome: TerminalOutcome; + reachedQueue: boolean; + answerDelaySeconds?: number; + talkTimeSeconds?: number; + hangupCause?: string; + agentId?: string; +} + +const OUTCOME_TO_STATUS: Record = { + COMPLETED: "COMPLETED", + BUSY: "BUSY", + NO_ANSWER: "NO_ANSWER", + FAILED: "FAILED", + ABANDONED: "ABANDONED", +}; + +/** + * Fecha uma tentativa: grava o CallAttempt terminal, atualiza a EWMA da + * campanha (agente.md secao 75) e decide o próximo passo do Lead — nunca + * retry infinito (secao 86): `attemptCount >= maxAttempts` vira + * MAX_ATTEMPTS, terminal, nunca mais selecionado. Leads que ainda vão + * tentar de novo voltam pra READY (não BUSY/NO_ANSWER/etc como status + * consultável — a razão especifica fica em `lastResult`; READY é o único + * jeito da query de reserva achar o lead de novo depois de + * `nextAttemptAt`). + */ +export async function completeAttempt( + tx: Prisma.TransactionClient, + params: CompleteAttemptParams, + campaignMaxAttempts: number, +): Promise { + const now = new Date(); + + await tx.callAttempt.update({ + where: { id: params.attemptId }, + data: { + status: OUTCOME_TO_STATUS[params.outcome], + talkTimeSeconds: params.talkTimeSeconds, + hangupCause: params.hangupCause, + agentId: params.agentId, + endedAt: now, + }, + }); + + const stats = await tx.campaignStats.findUniqueOrThrow({ where: { campaignId: params.campaignId } }); + const wasAnswered = params.outcome === "COMPLETED" || params.outcome === "ABANDONED"; + const data: Prisma.CampaignStatsUpdateInput = { + answerProbability: ewmaUpdate(stats.answerProbability, wasAnswered ? 1 : 0), + }; + if (wasAnswered && params.answerDelaySeconds != null) { + data.averageAnswerDelay = ewmaUpdate(stats.averageAnswerDelay, params.answerDelaySeconds); + } + if (params.talkTimeSeconds != null) { + data.averageTalkTime = ewmaUpdate(stats.averageTalkTime, params.talkTimeSeconds); + } + if (params.reachedQueue) { + data.abandonRate = ewmaUpdate(stats.abandonRate, params.outcome === "ABANDONED" ? 1 : 0); + } + await tx.campaignStats.update({ where: { campaignId: params.campaignId }, data }); + + const lead = await tx.lead.findUniqueOrThrow({ where: { id: params.leadId } }); + const attemptCount = lead.attemptCount + 1; + + if (params.outcome === "COMPLETED") { + await tx.lead.update({ + where: { id: params.leadId }, + data: { status: "COMPLETED", attemptCount, lastAttemptAt: now, lastResult: params.outcome }, + }); + return; + } + + if (attemptCount >= campaignMaxAttempts) { + await tx.lead.update({ + where: { id: params.leadId }, + data: { status: "MAX_ATTEMPTS", attemptCount, lastAttemptAt: now, lastResult: params.outcome }, + }); + return; + } + + await tx.lead.update({ + where: { id: params.leadId }, + data: { + status: "READY", + attemptCount, + lastAttemptAt: now, + lastResult: params.outcome, + nextAttemptAt: computeNextAttemptAt(params.outcome, now), + }, + }); +} diff --git a/apps/predictive-dialer/src/event-listener.ts b/apps/predictive-dialer/src/event-listener.ts new file mode 100644 index 0000000..8099043 --- /dev/null +++ b/apps/predictive-dialer/src/event-listener.ts @@ -0,0 +1,140 @@ +import Redis from "ioredis"; +import { getPrismaClient, withTenantContext, type Prisma } from "@b2bcall/database"; +import type { NormalizedEvent } from "@b2bcall/telephony"; +import { createLogger } from "@b2bcall/shared"; +import { completeAttempt } from "./call-attempt"; +import { getQueuedAttempt, unregisterQueuedAttempt } from "./queued-attempts-registry"; + +const logger = createLogger("b2bcall-predictive-dialer"); + +const EVENTS_CHANNEL = "b2bcall:events"; + +function extractId(fsName: string | undefined): string | undefined { + return fsName?.split("@")[0]; +} + +/** + * Assina o mesmo canal `b2bcall:events` que a fase Realtime Monitoring já + * usa (agente.md secao 24) — a perna sintética/real que entrou na fila + * carrega os channel variables b2bcall_* (secao 81), então os eventos dela + * chegam tenant-scoped de verdade, sem precisar do fan-out usado por + * eventos que não têm esses vars (ver docs/AGENTS.md, docs/REALTIME.md). + */ +export function startEventListener(redisUrl: string): Redis { + const subscriber = new Redis(redisUrl); + subscriber.on("error", (err) => logger.error("erro na conexao Redis (event listener)", { error: String(err) })); + subscriber.subscribe(EVENTS_CHANNEL).catch((err) => { + logger.error("falha ao assinar b2bcall:events", { error: String(err) }); + }); + + subscriber.on("message", (_channel, raw) => { + handleMessage(raw).catch((err) => { + logger.error("falha ao processar evento", { error: String(err) }); + }); + }); + + return subscriber; +} + +async function handleMessage(raw: string): Promise { + let event: NormalizedEvent; + try { + event = JSON.parse(raw); + } catch { + return; + } + + switch (event.type) { + case "CALL_ANSWERED": { + // Só relevante pro caminho PSTN real (secao 80-83) — a perna + // simulada já entra na fila com o delay pré-sorteado, sem passar por + // aqui (ver tick.ts). Nunca exercitado nesta sessão. + if (!event.callUuid) return; + const pending = getQueuedAttempt(event.callUuid); + if (!pending || pending.answerDelaySeconds != null) return; + pending.answeredAtMs = Date.now(); + return; + } + + case "AGENT_OFFERED_CALL": { + const memberSessionUuid = event.data.memberSessionUuid as string | undefined; + const agentFsName = event.data.agent as string | undefined; + const pending = memberSessionUuid && getQueuedAttempt(memberSessionUuid); + if (pending && agentFsName) { + pending.agentId = extractId(agentFsName); + } + return; + } + + case "CALL_BRIDGED": { + if (!event.callUuid) return; + const pending = getQueuedAttempt(event.callUuid); + if (!pending) return; + pending.bridgedAtMs = Date.now(); + const prisma = getPrismaClient(); + await withTenantContext(prisma, pending.tenantId, (tx) => + tx.callAttempt.update({ + where: { id: pending.attemptId }, + data: { status: "AGENT_CONNECTED", bridgedAt: new Date(), agentId: pending.agentId }, + }), + ); + return; + } + + case "QUEUE_MEMBER_LEFT": { + const memberSessionUuid = event.data.memberSessionUuid as string | undefined; + if (!memberSessionUuid) return; + const pending = getQueuedAttempt(memberSessionUuid); + if (!pending || pending.bridgedAtMs) return; // já foi pra um agente, quem fecha é CALL_ENDED + await finishAttempt(memberSessionUuid, pending, "ABANDONED"); + return; + } + + case "CALL_ENDED": { + if (!event.callUuid) return; + const pending = getQueuedAttempt(event.callUuid); + if (!pending) return; + const outcome = pending.bridgedAtMs ? "COMPLETED" : "ABANDONED"; + await finishAttempt(event.callUuid, pending, outcome, event.data.hangupCause as string | undefined); + return; + } + + default: + return; + } +} + +async function finishAttempt( + originationUuid: string, + pending: NonNullable>, + outcome: "COMPLETED" | "ABANDONED", + hangupCause?: string, +): Promise { + unregisterQueuedAttempt(originationUuid); + const prisma = getPrismaClient(); + const talkTimeSeconds = pending.bridgedAtMs ? Math.round((Date.now() - pending.bridgedAtMs) / 1000) : undefined; + const answerDelaySeconds = + pending.answerDelaySeconds ?? + (pending.answeredAtMs ? (pending.answeredAtMs - pending.queuedAtMs) / 1000 : undefined); + + await withTenantContext(prisma, pending.tenantId, (tx: Prisma.TransactionClient) => + completeAttempt( + tx, + { + attemptId: pending.attemptId, + tenantId: pending.tenantId, + campaignId: pending.campaignId, + leadId: pending.leadId, + outcome, + reachedQueue: true, + answerDelaySeconds, + talkTimeSeconds, + hangupCause, + agentId: pending.agentId, + }, + pending.maxAttempts, + ), + ); + + logger.info("tentativa finalizada (fila real)", { attemptId: pending.attemptId, outcome }); +} diff --git a/apps/predictive-dialer/src/ewma.ts b/apps/predictive-dialer/src/ewma.ts new file mode 100644 index 0000000..63bcd9c --- /dev/null +++ b/apps/predictive-dialer/src/ewma.ts @@ -0,0 +1,12 @@ +/** + * EWMA (agente.md secao 75) pra suavizar answer_probability/ + * average_answer_delay/TMA/abandon_rate — evita que o pacing oscile + * violentamente por causa de uma única amostra ruim. alpha alto = reage + * rápido a mudanças recentes; 0.25 é um meio-termo razoável (equivalente a + * uma janela de ~7-8 amostras). + */ +const ALPHA = 0.25; + +export function ewmaUpdate(previous: number, sample: number, alpha = ALPHA): number { + return alpha * sample + (1 - alpha) * previous; +} diff --git a/apps/predictive-dialer/src/lead-reservation.ts b/apps/predictive-dialer/src/lead-reservation.ts new file mode 100644 index 0000000..bbebee2 --- /dev/null +++ b/apps/predictive-dialer/src/lead-reservation.ts @@ -0,0 +1,51 @@ +import type { Prisma } from "@b2bcall/database"; + +export interface ReservedLead { + id: string; + phoneNormalized: string; + name: string | null; + attemptCount: number; +} + +/** + * Reserva atômica de leads (agente.md secao 78): `FOR UPDATE SKIP LOCKED` + * garante que dois workers concorrentes nunca pegam o mesmo lead — quem + * chegar primeiro tranca a linha, o outro pula pra próxima em vez de + * esperar (nada de fila de espera aqui, é melhor originar outro lead do + * que travar o tick inteiro). `count` normalmente é pequeno (poucas + * unidades por tick), então `SELECT ... LIMIT` é barato mesmo sem índice + * dedicado além do já existente em `(campaign_id, status, next_attempt_at)`. + * + * Precisa rodar dentro do MESMO `withTenantContext` transaction que fez a + * checagem de RLS — o lock de linha só vale até o commit/rollback da + * transação atual. + */ +export async function reserveLeads( + tx: Prisma.TransactionClient, + tenantId: string, + campaignId: string, + count: number, +): Promise { + if (count <= 0) return []; + + const rows = await tx.$queryRaw` + SELECT id, phone_normalized AS "phoneNormalized", name, attempt_count AS "attemptCount" + FROM leads + WHERE tenant_id = ${tenantId}::uuid + AND campaign_id = ${campaignId}::uuid + AND status IN ('NEW', 'READY') + AND (next_attempt_at IS NULL OR next_attempt_at <= now()) + ORDER BY next_attempt_at ASC NULLS FIRST, created_at ASC + LIMIT ${count} + FOR UPDATE SKIP LOCKED + `; + + if (rows.length === 0) return []; + + await tx.lead.updateMany({ + where: { id: { in: rows.map((r) => r.id) } }, + data: { status: "RESERVED" }, + }); + + return rows; +} diff --git a/apps/predictive-dialer/src/main.ts b/apps/predictive-dialer/src/main.ts new file mode 100644 index 0000000..33eab5e --- /dev/null +++ b/apps/predictive-dialer/src/main.ts @@ -0,0 +1,134 @@ +import { randomUUID } from "node:crypto"; +import Redis from "ioredis"; +import { getPrismaClient, withTenantContext } from "@b2bcall/database"; +import { FreeSwitchTelephonyProvider } from "@b2bcall/telephony"; +import { createLogger } from "@b2bcall/shared"; +import { acquireCampaignLock, renewCampaignLock, releaseCampaignLock } from "./redis-primitives"; +import { tickCampaign, type TickDeps } from "./tick"; +import { startEventListener } from "./event-listener"; + +const logger = createLogger("b2bcall-predictive-dialer"); + +const TICK_INTERVAL_MS = 2000; +const LOCK_TTL_MS = 10_000; + +function requireEnv(name: string): string { + const value = process.env[name]; + if (!value) throw new Error(`${name} nao definido no ambiente`); + return value; +} + +/** + * Real Outbound Safety (agente.md secao 186): as DUAS condições precisam + * estar explicitamente ligadas — nunca ativado automaticamente. Lido uma + * vez no boot (não é algo que deveria mudar em runtime sem reiniciar o + * worker). + */ +function readOutboundSafetyFlags(): { dialerSimulation: boolean; allowRealOutboundCalls: boolean } { + const dialerSimulation = (process.env.DIALER_SIMULATION ?? "true") !== "false"; + const allowRealOutboundCalls = process.env.ALLOW_REAL_OUTBOUND_CALLS === "true"; + if (dialerSimulation) { + logger.info("DIALER_SIMULATION=true — nenhuma chamada PSTN real sera originada"); + } else if (!allowRealOutboundCalls) { + logger.warn( + "DIALER_SIMULATION=false mas ALLOW_REAL_OUTBOUND_CALLS != true — chamadas reais continuam bloqueadas (secao 186)", + ); + } else { + logger.warn("CHAMADAS PSTN REAIS HABILITADAS (DIALER_SIMULATION=false + ALLOW_REAL_OUTBOUND_CALLS=true)"); + } + return { dialerSimulation, allowRealOutboundCalls }; +} + +async function main(): Promise { + const { dialerSimulation, allowRealOutboundCalls } = readOutboundSafetyFlags(); + const redis = new Redis(requireEnv("REDIS_URL")); + redis.on("error", (err) => logger.error("erro na conexao Redis", { error: String(err) })); + + const provider = new FreeSwitchTelephonyProvider({ + host: requireEnv("ESL_HOST"), + port: Number(process.env.ESL_PORT ?? 8021), + password: requireEnv("ESL_PASSWORD"), + logger: { + debug: () => {}, + info: (msg) => logger.debug(msg), + error: (msg, data) => logger.error(msg, { detail: data }), + }, + }); + provider.connect(); + await provider.waitUntilConnected(10_000); + + const eventSubscriber = startEventListener(requireEnv("REDIS_URL")); + + const workerId = randomUUID(); + const deps: TickDeps = { redis, provider, dialerSimulation, allowRealOutboundCalls }; + + let stopped = false; + const tick = async () => { + if (stopped) return; + try { + await runTick(deps, workerId); + } catch (err) { + logger.error("falha no tick", { error: String(err) }); + } + if (!stopped) setTimeout(tick, TICK_INTERVAL_MS); + }; + setTimeout(tick, TICK_INTERVAL_MS); + + logger.info("b2bcall-predictive-dialer iniciado", { workerId, tickIntervalMs: TICK_INTERVAL_MS }); + + const shutdown = async () => { + stopped = true; + logger.info("encerrando b2bcall-predictive-dialer"); + await provider.disconnect(); + redis.disconnect(); + eventSubscriber.disconnect(); + process.exit(0); + }; + process.on("SIGTERM", shutdown); + process.on("SIGINT", shutdown); +} + +/** + * Um tick = uma passada por todos os tenants ativos, campanhas RUNNING (ou + * WAITING_SCHEDULE, que pode voltar a RUNNING dentro da janela — ver + * tick.ts). Lock por campanha (secao 79) garante que, mesmo rodando mais + * de um worker deste serviço, só um processa uma dada campanha por vez — + * renovado durante o processamento, liberado com compare-and-delete no + * final (nunca libera o lock de outro dono). + */ +async function runTick(deps: TickDeps, workerId: string): Promise { + const prisma = getPrismaClient(); + const tenants = await prisma.tenant.findMany({ where: { status: "ACTIVE" } }); + + for (const tenant of tenants) { + const campaigns = await withTenantContext(prisma, tenant.id, (tx) => + tx.campaign.findMany({ + where: { tenantId: tenant.id, deletedAt: null, status: { in: ["RUNNING", "WAITING_SCHEDULE"] } }, + }), + ); + + for (const campaign of campaigns) { + const ownerToken = `${workerId}:${randomUUID()}`; + const locked = await acquireCampaignLock(deps.redis, campaign.id, ownerToken, LOCK_TTL_MS); + if (!locked) continue; + + const renewTimer = setInterval(() => { + renewCampaignLock(deps.redis, campaign.id, ownerToken, LOCK_TTL_MS).catch(() => undefined); + }, LOCK_TTL_MS / 2); + + try { + await tickCampaign(deps, tenant, campaign); + } catch (err) { + logger.error("falha no tick da campanha", { error: String(err), campaignId: campaign.id }); + } finally { + clearInterval(renewTimer); + await releaseCampaignLock(deps.redis, campaign.id, ownerToken); + } + } + } +} + +main().catch((err) => { + logger.error("falha fatal ao iniciar b2bcall-predictive-dialer", { error: String(err) }); + process.exit(1); +}); diff --git a/apps/predictive-dialer/src/originate.ts b/apps/predictive-dialer/src/originate.ts new file mode 100644 index 0000000..6aecf3f --- /dev/null +++ b/apps/predictive-dialer/src/originate.ts @@ -0,0 +1,75 @@ +import type { FreeSwitchTelephonyProvider } from "@b2bcall/telephony"; + +export interface OriginateIdentifiers { + tenantId: string; + attemptId: string; + campaignId: string; + leadId: string; +} + +function channelVars(ids: OriginateIdentifiers): Record { + // agente.md secao 81: tenant_id/call_id/attempt_id/campaign_id/lead_id + // como channel variables. Não existe uma entidade "Call" separada de + // CallAttempt nesta fase — b2bcall_call_id usa o mesmo id do attempt. + return { + b2bcall_tenant_id: ids.tenantId, + b2bcall_call_id: ids.attemptId, + b2bcall_attempt_id: ids.attemptId, + b2bcall_campaign_id: ids.campaignId, + b2bcall_lead_id: ids.leadId, + }; +} + +/** + * Perna "atendida" sintética (modo simulação): quando o outcome sorteado é + * ANSWERED, origina uma chamada de verdade no FreeSWITCH (`null/dummy`, + * sem PSTN nenhum envolvido) direto pra `&callcenter(...)` — a parte + * "cliente atendeu" é simulada, mas a partir daqui o pipeline de + * fila/agente é 100% real (o mesmo mod_callcenter já testado nas fases + * Queues/Agents/Realtime Monitoring). + */ +export async function originateSimulatedAnswerLeg( + provider: FreeSwitchTelephonyProvider, + ids: OriginateIdentifiers, + queueId: string, + domain: string, +): Promise<{ uuid: string }> { + return provider.originate({ + destination: "null/dummy", + application: "callcenter", + applicationArgs: `${queueId}@${domain}`, + channelVariables: channelVars(ids), + }); +} + +/** + * Perna PSTN real (agente.md secao 80-83) — `bgapi originate` via Sofia + * Gateway. Só é chamada quando `DIALER_SIMULATION=false` E + * `ALLOW_REAL_OUTBOUND_CALLS=true` (secao 186, checado em main.ts antes de + * sequer construir o tick). **Nunca exercitada nesta sessão** — não há + * trunk/operadora real disponível neste laboratório; ver + * docs/PREDICTIVE_DIALER.md. + */ +export async function originateRealPstnLeg( + provider: FreeSwitchTelephonyProvider, + ids: OriginateIdentifiers, + params: { + trunkId: string; + phoneNumber: string; + queueId: string; + domain: string; + callerIdName?: string; + callerIdNumber?: string; + ringTimeoutSeconds: number; + }, +): Promise<{ uuid: string }> { + return provider.originate({ + destination: `sofia/gateway/${params.trunkId}/${params.phoneNumber}`, + application: "callcenter", + applicationArgs: `${params.queueId}@${params.domain}`, + channelVariables: channelVars(ids), + callerIdName: params.callerIdName, + callerIdNumber: params.callerIdNumber, + timeoutSeconds: params.ringTimeoutSeconds, + }); +} diff --git a/apps/predictive-dialer/src/pacing.ts b/apps/predictive-dialer/src/pacing.ts new file mode 100644 index 0000000..6617a09 --- /dev/null +++ b/apps/predictive-dialer/src/pacing.ts @@ -0,0 +1,155 @@ +import type { Campaign, CampaignStats, Prisma } from "@b2bcall/database"; + +/** + * Dados em tempo real considerados pelo motor (agente.md secao 73) — + * derivados na hora a partir de Agent.state (fonte de verdade já mantida + * pelas fases Agents/Realtime Monitoring), não persistidos separadamente. + */ +export interface CapacitySnapshot { + availableAgents: number; + reservedAgents: number; + ringingAgents: number; + agentsInCall: number; + agentsInWrapup: number; + agentsPaused: number; + predictedBecomingAvailable: number; + callsInFlight: number; +} + +const PREDICTION_HORIZON_SECONDS = 15; + +/** + * Previsão de liberação (agente.md secao 74) — versão determinística e + * explicável, sem Machine Learning: pra cada agente em IN_CALL/WRAP_UP, + * estima o tempo restante (average_talk_time ou wrap_up_time menos o + * tempo já decorrido desde `stateUpdatedAt`) e conta quantos cruzam o + * horizonte de previsão. Simplificação deliberada da tabela de 4 buckets + * (5/10/15/20s) da especificação — um único horizonte de 15s, documentado + * em docs/PREDICTIVE_DIALER.md; refinar pros 4 buckets é trabalho futuro. + */ +export async function computeCapacity( + tx: Prisma.TransactionClient, + tenantId: string, + campaign: Campaign, + stats: CampaignStats, +): Promise { + const agents = await tx.agent.findMany({ + where: { tenantId, enabled: true, deletedAt: null, tiers: { some: { queueId: campaign.queueId } } }, + select: { state: true, stateUpdatedAt: true, wrapUpTime: true }, + }); + + const now = Date.now(); + let availableAgents = 0; + let reservedAgents = 0; + let ringingAgents = 0; + let agentsInCall = 0; + let agentsInWrapup = 0; + let agentsPaused = 0; + let predictedBecomingAvailable = 0; + + for (const agent of agents) { + const elapsedSeconds = agent.stateUpdatedAt ? (now - agent.stateUpdatedAt.getTime()) / 1000 : Infinity; + + switch (agent.state) { + case "AVAILABLE": + availableAgents++; + break; + case "RESERVED": + reservedAgents++; + break; + case "RINGING": + ringingAgents++; + break; + case "IN_CALL": { + agentsInCall++; + const remaining = stats.averageTalkTime - elapsedSeconds; + if (remaining <= PREDICTION_HORIZON_SECONDS) predictedBecomingAvailable++; + break; + } + case "WRAP_UP": { + agentsInWrapup++; + const remaining = agent.wrapUpTime - elapsedSeconds; + if (remaining <= PREDICTION_HORIZON_SECONDS) predictedBecomingAvailable++; + break; + } + case "PAUSED": + agentsPaused++; + break; + default: + break; + } + } + + const callsInFlight = await tx.callAttempt.count({ + where: { + tenantId, + campaignId: campaign.id, + status: { in: ["CREATED", "RESERVED", "ORIGINATING", "ORIGINATED", "RINGING", "ANSWERED", "QUEUEING"] }, + }, + }); + + return { + availableAgents, + reservedAgents, + ringingAgents, + agentsInCall, + agentsInWrapup, + agentsPaused, + predictedBecomingAvailable, + callsInFlight, + }; +} + +export interface PacingDecision { + callsToOriginate: number; + newPacingFactor: number; +} + +/** + * Cálculo conceitual (agente.md secao 76): expected_agent_capacity = + * available + predicted_becoming_available; expected_answers = calls * + * answer_probability; objetivo expected_answers ≈ expected_agent_capacity, + * aplicando pacing_factor e limites. Controle de abandono (secao 84): + * abandon_rate acima do alvo reduz o pacing; acima de 2x entra em modo + * conservador (pacing mínimo); acima de 3x suspende originações neste + * tick. Nunca discar sem capacidade prevista (secao 85). + */ +export function decidePacing( + campaign: Campaign, + stats: CampaignStats, + capacity: CapacitySnapshot, +): PacingDecision { + let pacingFactor = stats.pacingFactor; + + if (stats.abandonRate > campaign.targetAbandonRate * 3) { + // Modo crítico: suspende originações neste tick, mas não zera o + // pacingFactor guardado (evita ficar preso no mínimo por uma amostra + // ruim isolada — a próxima leitura de abandonRate decide de novo). + return { callsToOriginate: 0, newPacingFactor: campaign.pacingMin }; + } + if (stats.abandonRate > campaign.targetAbandonRate * 2) { + pacingFactor = campaign.pacingMin; + } else if (stats.abandonRate > campaign.targetAbandonRate) { + pacingFactor = Math.max(campaign.pacingMin, pacingFactor * 0.8); + } else if (pacingFactor < campaign.pacingMax) { + pacingFactor = Math.min(campaign.pacingMax, pacingFactor * 1.05); + } + + const expectedAgentCapacity = capacity.availableAgents + capacity.predictedBecomingAvailable; + if (expectedAgentCapacity <= 0) { + // Secao 85: não originar agressivamente sem capacidade prevista. + return { callsToOriginate: 0, newPacingFactor: pacingFactor }; + } + + const targetOriginations = expectedAgentCapacity * pacingFactor; + const answerProbability = Math.max(0.01, stats.answerProbability); + let callsToOriginate = Math.round(targetOriginations / answerProbability) - capacity.callsInFlight; + callsToOriginate = Math.max(0, callsToOriginate); + + if (campaign.maxConcurrentCalls != null) { + const room = campaign.maxConcurrentCalls - capacity.callsInFlight; + callsToOriginate = Math.min(callsToOriginate, Math.max(0, room)); + } + + return { callsToOriginate, newPacingFactor: pacingFactor }; +} diff --git a/apps/predictive-dialer/src/queued-attempts-registry.ts b/apps/predictive-dialer/src/queued-attempts-registry.ts new file mode 100644 index 0000000..47be443 --- /dev/null +++ b/apps/predictive-dialer/src/queued-attempts-registry.ts @@ -0,0 +1,44 @@ +/** + * Correlaciona eventos reais do FreeSWITCH (chegando via Redis + * `b2bcall:events`, publicados por b2bcall-fs-events) de volta pra um + * CallAttempt em andamento — registry em memória, chave é o + * `origination_uuid` da perna sintética (simulada) ou real que entrou na + * fila. Não precisa persistir: se o worker reiniciar no meio de uma + * chamada em curso, essa chamada específica fica "órfã" (perde a + * correlação, mas o CallAttempt já teria os timestamps que existiam até + * ali) — aceitável nesta fase, mesma classe de limitação já documentada + * pra outros processos de curta duração deste projeto. + */ +export interface QueuedAttemptInfo { + attemptId: string; + tenantId: string; + campaignId: string; + leadId: string; + maxAttempts: number; + queuedAtMs: number; + // Conhecido de antemão no modo simulação (o delay já foi sorteado antes + // de originar); no caminho PSTN real, fica undefined até answeredAtMs + // ser preenchido pelo evento CALL_ANSWERED real. + answerDelaySeconds?: number; + answeredAtMs?: number; + agentId?: string; + bridgedAtMs?: number; +} + +const registry = new Map(); + +export function registerQueuedAttempt(originationUuid: string, info: QueuedAttemptInfo): void { + registry.set(originationUuid, info); +} + +export function getQueuedAttempt(originationUuid: string): QueuedAttemptInfo | undefined { + return registry.get(originationUuid); +} + +export function unregisterQueuedAttempt(originationUuid: string): void { + registry.delete(originationUuid); +} + +export function pendingCount(): number { + return registry.size; +} diff --git a/apps/predictive-dialer/src/redis-primitives.ts b/apps/predictive-dialer/src/redis-primitives.ts new file mode 100644 index 0000000..1ddacfb --- /dev/null +++ b/apps/predictive-dialer/src/redis-primitives.ts @@ -0,0 +1,98 @@ +import type Redis from "ioredis"; + +/** + * CPS distribuído (agente.md secao 77): "token bucket ou equivalente, + * Redis, múltiplos workers, nunca sleep() como controle." Implementado + * como um contador de janela fixa de 1s por chave (INCR + PEXPIRE), checado + * e incrementado num único script Lua — atômico mesmo com N workers + * concorrentes, sem sleep nenhum. + * + * A hierarquia (agente.md secao 62: GLOBAL → NODE → TENANT → TRUNK → + * CAMPAIGN) é uma única chamada com várias chaves: só incrementa TODAS se + * TODAS tiverem espaço — nunca incrementa parcialmente e desfaz depois. + */ +const CPS_WINDOW_SCRIPT = ` +local n = #KEYS +for i = 1, n do + local limit = tonumber(ARGV[i]) + local current = tonumber(redis.call('GET', KEYS[i]) or '0') + if current >= limit then + return 0 + end +end +for i = 1, n do + local newval = redis.call('INCR', KEYS[i]) + if newval == 1 then + redis.call('PEXPIRE', KEYS[i], 1000) + end +end +return 1 +`; + +export interface CpsCheck { + key: string; + maxPerSecond: number; +} + +/** Tenta consumir 1 slot de CPS em TODOS os níveis da hierarquia de uma vez + * (só chaves com limite definido — null/undefined = sem limite nesse + * nível, nem entra no script). Retorna false se qualquer nível estourar. */ +export async function tryAcquireCps(redis: Redis, checks: CpsCheck[]): Promise { + const applicable = checks.filter((c) => c.maxPerSecond > 0); + if (applicable.length === 0) return true; + const keys = applicable.map((c) => c.key); + const args = applicable.map((c) => String(c.maxPerSecond)); + const result = await redis.eval(CPS_WINDOW_SCRIPT, keys.length, ...keys, ...args); + return result === 1; +} + +/** + * Lock de campanha (agente.md secao 79): `dialer:campaign:{id}`, TTL, + * ownership token (só quem detém o token renova/libera), renewal, safe + * release (compare-and-delete via Lua — nunca libera o lock de outro dono + * por engano numa corrida entre "TTL expirou" e "release chegou atrasado"). + */ +function lockKey(campaignId: string): string { + return `dialer:campaign:${campaignId}`; +} + +export async function acquireCampaignLock( + redis: Redis, + campaignId: string, + ownerToken: string, + ttlMs: number, +): Promise { + const result = await redis.set(lockKey(campaignId), ownerToken, "PX", ttlMs, "NX"); + return result === "OK"; +} + +const RENEW_SCRIPT = ` +if redis.call('GET', KEYS[1]) == ARGV[1] then + return redis.call('PEXPIRE', KEYS[1], ARGV[2]) +else + return 0 +end +`; + +export async function renewCampaignLock( + redis: Redis, + campaignId: string, + ownerToken: string, + ttlMs: number, +): Promise { + const result = await redis.eval(RENEW_SCRIPT, 1, lockKey(campaignId), ownerToken, String(ttlMs)); + return result === 1; +} + +const RELEASE_SCRIPT = ` +if redis.call('GET', KEYS[1]) == ARGV[1] then + return redis.call('DEL', KEYS[1]) +else + return 0 +end +`; + +export async function releaseCampaignLock(redis: Redis, campaignId: string, ownerToken: string): Promise { + const result = await redis.eval(RELEASE_SCRIPT, 1, lockKey(campaignId), ownerToken); + return result === 1; +} diff --git a/apps/predictive-dialer/src/retry-rules.ts b/apps/predictive-dialer/src/retry-rules.ts new file mode 100644 index 0000000..b4bc270 --- /dev/null +++ b/apps/predictive-dialer/src/retry-rules.ts @@ -0,0 +1,16 @@ +/** + * Regras de retry configuráveis (agente.md secao 86) — nunca retry + * infinito, `Campaign.maxAttempts` é o teto absoluto (aplicado por quem + * chama, ver call-attempt.ts). + */ +const RETRY_DELAY_MINUTES: Record = { + BUSY: 15, + NO_ANSWER: 60, + FAILED: 30, + ABANDONED: 30, +}; + +export function computeNextAttemptAt(outcome: keyof typeof RETRY_DELAY_MINUTES, now: Date): Date { + const minutes = RETRY_DELAY_MINUTES[outcome] ?? 60; + return new Date(now.getTime() + minutes * 60_000); +} diff --git a/apps/predictive-dialer/src/schedule.ts b/apps/predictive-dialer/src/schedule.ts new file mode 100644 index 0000000..3bc6d55 --- /dev/null +++ b/apps/predictive-dialer/src/schedule.ts @@ -0,0 +1,43 @@ +import type { Campaign } from "@b2bcall/database"; + +/** + * Janela de funcionamento da campanha (agente.md secao 63-64: timezone, + * days_of_week, start/end_date, start/end_time — status vira + * WAITING_SCHEDULE quando fora da janela). Usa `Intl.DateTimeFormat` com + * `timeZone` em vez de trazer uma lib de datas só pra isso — o Node já sabe + * converter pra qualquer timezone IANA nativamente. + */ +export function isWithinSchedule(campaign: Campaign, now: Date = new Date()): boolean { + const parts = new Intl.DateTimeFormat("en-US", { + timeZone: campaign.timezone, + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + hour12: false, + weekday: "short", + }).formatToParts(now); + + const get = (type: string) => parts.find((p) => p.type === type)?.value ?? ""; + const dateStr = `${get("year")}-${get("month")}-${get("day")}`; + const timeStr = `${get("hour")}:${get("minute")}`; + const weekdayMap: Record = { Mon: 1, Tue: 2, Wed: 3, Thu: 4, Fri: 5, Sat: 6, Sun: 7 }; + const isoWeekday = weekdayMap[get("weekday")]; + + if (campaign.startDate) { + const startStr = campaign.startDate.toISOString().slice(0, 10); + if (dateStr < startStr) return false; + } + if (campaign.endDate) { + const endStr = campaign.endDate.toISOString().slice(0, 10); + if (dateStr > endStr) return false; + } + if (campaign.daysOfWeek.length > 0 && !campaign.daysOfWeek.includes(isoWeekday)) { + return false; + } + if (campaign.startTime && timeStr < campaign.startTime) return false; + if (campaign.endTime && timeStr > campaign.endTime) return false; + + return true; +} diff --git a/apps/predictive-dialer/src/simulation.ts b/apps/predictive-dialer/src/simulation.ts new file mode 100644 index 0000000..e0e26aa --- /dev/null +++ b/apps/predictive-dialer/src/simulation.ts @@ -0,0 +1,53 @@ +/** + * Modo simulação (agente.md secao 185): `DIALER_SIMULATION=true` (default) + * — nenhuma chamada PSTN real acontece. Simula ANSWER/BUSY/NO_ANSWER/ + * FAILED/RINGING com delay e talk time sorteados, tudo em software, sem + * originar nada pro trunk real. Habilitar chamadas reais exige as DUAS + * condições da secao 186 (`DIALER_SIMULATION=false` E + * `ALLOW_REAL_OUTBOUND_CALLS=true`) — nunca ativado automaticamente, ver + * main.ts. + */ +export type SimulatedOutcomeType = "ANSWERED" | "BUSY" | "NO_ANSWER" | "FAILED"; + +export interface SimulatedOutcome { + type: SimulatedOutcomeType; + ringDelayMs: number; + talkTimeSeconds?: number; +} + +const OUTCOME_PROBABILITIES: [SimulatedOutcomeType, number][] = [ + ["ANSWERED", 0.4], + ["BUSY", 0.15], + ["NO_ANSWER", 0.35], + ["FAILED", 0.1], +]; + +function randomBetween(min: number, max: number): number { + return min + Math.random() * (max - min); +} + +export function simulateOutcome(ringTimeoutSeconds: number, averageTalkTimeSeconds: number): SimulatedOutcome { + const roll = Math.random(); + let cumulative = 0; + let type: SimulatedOutcomeType = "FAILED"; + for (const [candidate, probability] of OUTCOME_PROBABILITIES) { + cumulative += probability; + if (roll <= cumulative) { + type = candidate; + break; + } + } + + if (type === "ANSWERED") { + const ringDelayMs = randomBetween(1000, Math.min(8000, ringTimeoutSeconds * 1000 * 0.6)); + // Talk time varia +-40% em torno da média corrente da campanha (EWMA) — + // sem distribuição estatística real por trás (não existe histórico de + // chamadas de verdade ainda), só o suficiente pra exercitar o pipeline + // com valores plausíveis. + const talkTimeSeconds = Math.max(10, randomBetween(averageTalkTimeSeconds * 0.6, averageTalkTimeSeconds * 1.4)); + return { type, ringDelayMs, talkTimeSeconds }; + } + + const ringDelayMs = randomBetween(1000, ringTimeoutSeconds * 1000); + return { type, ringDelayMs }; +} diff --git a/apps/predictive-dialer/src/tick.ts b/apps/predictive-dialer/src/tick.ts new file mode 100644 index 0000000..05a0ea8 --- /dev/null +++ b/apps/predictive-dialer/src/tick.ts @@ -0,0 +1,244 @@ +import type Redis from "ioredis"; +import { getPrismaClient, withTenantContext, type Campaign, type Tenant } from "@b2bcall/database"; +import type { FreeSwitchTelephonyProvider } from "@b2bcall/telephony"; +import { createLogger } from "@b2bcall/shared"; +import { tryAcquireCps } from "./redis-primitives"; +import { reserveLeads } from "./lead-reservation"; +import { computeCapacity, decidePacing } from "./pacing"; +import { isWithinSchedule } from "./schedule"; +import { simulateOutcome } from "./simulation"; +import { originateSimulatedAnswerLeg, originateRealPstnLeg } from "./originate"; +import { createCallAttempt, setAttemptStatus, completeAttempt } from "./call-attempt"; +import { registerQueuedAttempt } from "./queued-attempts-registry"; + +const logger = createLogger("b2bcall-predictive-dialer"); + +export interface TickDeps { + redis: Redis; + provider: FreeSwitchTelephonyProvider; + dialerSimulation: boolean; + allowRealOutboundCalls: boolean; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export async function tickCampaign(deps: TickDeps, tenant: Tenant, campaign: Campaign): Promise { + const prisma = getPrismaClient(); + + if (!isWithinSchedule(campaign)) { + if (campaign.status === "RUNNING") { + await withTenantContext(prisma, tenant.id, (tx) => + tx.campaign.update({ where: { id: campaign.id }, data: { status: "WAITING_SCHEDULE" } }), + ); + logger.info("campanha fora da janela de funcionamento", { campaignId: campaign.id }); + } + return; + } + if (campaign.status === "WAITING_SCHEDULE") { + await withTenantContext(prisma, tenant.id, (tx) => + tx.campaign.update({ where: { id: campaign.id }, data: { status: "RUNNING" } }), + ); + } + + const stats = await withTenantContext(prisma, tenant.id, (tx) => + tx.campaignStats.upsert({ + where: { campaignId: campaign.id }, + update: {}, + create: { campaignId: campaign.id, tenantId: tenant.id, pacingFactor: campaign.pacingInitial }, + }), + ); + + const capacity = await withTenantContext(prisma, tenant.id, (tx) => computeCapacity(tx, tenant.id, campaign, stats)); + const { callsToOriginate, newPacingFactor } = decidePacing(campaign, stats, capacity); + + if (newPacingFactor !== stats.pacingFactor) { + await withTenantContext(prisma, tenant.id, (tx) => + tx.campaignStats.update({ where: { campaignId: campaign.id }, data: { pacingFactor: newPacingFactor } }), + ); + } + + if (callsToOriginate <= 0) return; + + const reserved = await withTenantContext(prisma, tenant.id, (tx) => + reserveLeads(tx, tenant.id, campaign.id, callsToOriginate), + ); + if (reserved.length === 0) return; + + logger.info("originando tentativas", { + campaignId: campaign.id, + count: reserved.length, + availableAgents: capacity.availableAgents, + predictedBecomingAvailable: capacity.predictedBecomingAvailable, + pacingFactor: newPacingFactor, + answerProbability: stats.answerProbability, + }); + + for (const lead of reserved) { + const cpsChecks = [ + { key: "cps:global", maxPerSecond: 0 }, // sem teto global configurado nesta fase + { key: `cps:tenant:${tenant.id}`, maxPerSecond: (await getTenantMaxCps(tenant.id)) ?? 0 }, + { key: `cps:campaign:${campaign.id}`, maxPerSecond: campaign.maxCps ?? 0 }, + { key: `cps:trunk:${campaign.trunkId}`, maxPerSecond: 0 }, // Trunk.maxCps já e' opcional; aplicado no real-outbound path + ]; + const allowed = await tryAcquireCps(deps.redis, cpsChecks); + if (!allowed) { + // De volta pra READY: essa reserva não gerou tentativa nenhuma, não + // conta como attempt (agente.md secao 62: hierarquia de CPS respeitada + // antes de originar, não depois). + await withTenantContext(prisma, tenant.id, (tx) => + tx.lead.update({ where: { id: lead.id }, data: { status: "READY" } }), + ); + continue; + } + + await originateOneAttempt(deps, tenant, campaign, lead.id); + } +} + +async function getTenantMaxCps(tenantId: string): Promise { + const prisma = getPrismaClient(); + const tenant = await prisma.tenant.findUniqueOrThrow({ where: { id: tenantId }, include: { plan: true } }); + return tenant.plan.maxCps; +} + +async function originateOneAttempt( + deps: TickDeps, + tenant: Tenant, + campaign: Campaign, + leadId: string, +): Promise { + const prisma = getPrismaClient(); + const simulated = deps.dialerSimulation || !deps.allowRealOutboundCalls; + + const attempt = await withTenantContext(prisma, tenant.id, (tx) => + createCallAttempt(tx, { tenantId: tenant.id, campaignId: campaign.id, leadId, simulated }), + ); + + if (simulated) { + runSimulatedAttempt(deps, tenant, campaign, leadId, attempt.id).catch((err) => { + logger.error("falha na simulacao da tentativa", { error: String(err), attemptId: attempt.id }); + }); + return; + } + + await runRealAttempt(deps, tenant, campaign, leadId, attempt.id); +} + +/** + * Modo simulação (agente.md secao 185): sorteia o desfecho da "chamada + * PSTN" inteiramente em software (RINGING -> ANSWER/BUSY/NO_ANSWER/FAILED + * com delay), sem tocar o FreeSWITCH pra isso. Só quando o sorteio dá + * ANSWERED é que uma chamada real (sintética, sem PSTN) entra na fila de + * verdade — a partir daí quem decide o resto é o mod_callcenter real, + * observado via event-listener.ts. + */ +async function runSimulatedAttempt( + deps: TickDeps, + tenant: Tenant, + campaign: Campaign, + leadId: string, + attemptId: string, +): Promise { + const prisma = getPrismaClient(); + const stats = await withTenantContext(prisma, tenant.id, (tx) => + tx.campaignStats.findUniqueOrThrow({ where: { campaignId: campaign.id } }), + ); + const outcome = simulateOutcome(campaign.ringTimeout, stats.averageTalkTime); + + await withTenantContext(prisma, tenant.id, (tx) => setAttemptStatus(tx, attemptId, "RINGING", { ringingAt: new Date() })); + await sleep(outcome.ringDelayMs); + + if (outcome.type !== "ANSWERED") { + await withTenantContext(prisma, tenant.id, (tx) => + completeAttempt( + tx, + { + attemptId, + tenantId: tenant.id, + campaignId: campaign.id, + leadId, + outcome: outcome.type as "BUSY" | "NO_ANSWER" | "FAILED", + reachedQueue: false, + }, + campaign.maxAttempts, + ), + ); + return; + } + + const answerDelaySeconds = outcome.ringDelayMs / 1000; + await withTenantContext(prisma, tenant.id, (tx) => setAttemptStatus(tx, attemptId, "ANSWERED", { answeredAt: new Date() })); + + const { uuid } = await originateSimulatedAnswerLeg( + deps.provider, + { tenantId: tenant.id, attemptId, campaignId: campaign.id, leadId }, + campaign.queueId, + tenant.telephonyDomain ?? "", + ); + + registerQueuedAttempt(uuid, { + attemptId, + tenantId: tenant.id, + campaignId: campaign.id, + leadId, + maxAttempts: campaign.maxAttempts, + queuedAtMs: Date.now(), + answerDelaySeconds, + }); + + await withTenantContext(prisma, tenant.id, (tx) => setAttemptStatus(tx, attemptId, "QUEUEING", { originationUuid: uuid })); + + // A perna sintética (null/dummy) não tem mídia real do outro lado — nada + // faz a chamada terminar sozinha depois de bridgear com um agente + // (diferente de uma ligação PSTN de verdade, onde o cliente desliga). + // Encerra explicitamente depois do talk_time simulado; se a chamada já + // tiver terminado antes disso (abandonada na fila, por exemplo), + // `killCall` num uuid que não existe mais só retorna erro, sem efeito. + setTimeout(() => { + deps.provider.killCall(uuid, "NORMAL_CLEARING").catch(() => undefined); + }, outcome.talkTimeSeconds! * 1000); +} + +/** + * Perna PSTN real (agente.md secao 80-83, 186) — só chamada quando as DUAS + * flags de segurança estão explicitamente ligadas. Nunca exercitada nesta + * sessão (sem trunk/operadora real disponível) — ver + * docs/PREDICTIVE_DIALER.md. + */ +async function runRealAttempt( + deps: TickDeps, + tenant: Tenant, + campaign: Campaign, + leadId: string, + attemptId: string, +): Promise { + const prisma = getPrismaClient(); + const lead = await withTenantContext(prisma, tenant.id, (tx) => tx.lead.findUniqueOrThrow({ where: { id: leadId } })); + + const { uuid } = await originateRealPstnLeg( + deps.provider, + { tenantId: tenant.id, attemptId, campaignId: campaign.id, leadId }, + { + trunkId: campaign.trunkId, + phoneNumber: lead.phoneNormalized, + queueId: campaign.queueId, + domain: tenant.telephonyDomain ?? "", + callerIdName: campaign.callerIdName ?? undefined, + callerIdNumber: campaign.callerIdNumber ?? undefined, + ringTimeoutSeconds: campaign.ringTimeout, + }, + ); + + registerQueuedAttempt(uuid, { + attemptId, + tenantId: tenant.id, + campaignId: campaign.id, + leadId, + maxAttempts: campaign.maxAttempts, + queuedAtMs: Date.now(), + }); + + await withTenantContext(prisma, tenant.id, (tx) => setAttemptStatus(tx, attemptId, "ORIGINATING", { originationUuid: uuid })); +} diff --git a/apps/predictive-dialer/tsconfig.json b/apps/predictive-dialer/tsconfig.json new file mode 100644 index 0000000..3fbff51 --- /dev/null +++ b/apps/predictive-dialer/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "dist", + "rootDir": "src", + "types": ["node"] + }, + "include": ["src"] +} diff --git a/docker-compose.yml b/docker-compose.yml index 2d9e4f9..5f4d6c0 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -117,6 +117,28 @@ services: # ainda roda no host) — ver docs/NETWORK_ARCHITECTURE.md. REDIS_URL: redis://:${REDIS_PASSWORD}@redis:6379 + predictive-dialer: + build: + context: . + dockerfile: apps/predictive-dialer/Dockerfile + container_name: b2bcall-predictive-dialer + restart: unless-stopped + depends_on: + - freeswitch + - redis + - postgres + environment: + ESL_HOST: freeswitch + ESL_PORT: "8021" + ESL_PASSWORD: ${ESL_PASSWORD} + APP_DATABASE_URL: postgresql://${POSTGRES_APP_USER}:${POSTGRES_APP_PASSWORD}@postgres:5432/${POSTGRES_DB}?schema=public + REDIS_URL: redis://:${REDIS_PASSWORD}@redis:6379 + # Real Outbound Safety (agente.md secao 186): as DUAS precisam estar + # explicitamente ligadas pra originar PSTN de verdade — nunca ativado + # por default. Ver docs/PREDICTIVE_DIALER.md. + DIALER_SIMULATION: ${DIALER_SIMULATION:-true} + ALLOW_REAL_OUTBOUND_CALLS: ${ALLOW_REAL_OUTBOUND_CALLS:-false} + secrets: freeswitch_pat: environment: FREESWITCH_PAT diff --git a/docs/CAMPAIGNS.md b/docs/CAMPAIGNS.md index 095c1fe..bb15cd2 100644 --- a/docs/CAMPAIGNS.md +++ b/docs/CAMPAIGNS.md @@ -107,13 +107,8 @@ delete numa campanha RUNNING -> 400 "Pare a campanha antes de apaga-la" ## O que falta -- `PredictiveDialerEngine` inteiro (secao 72-86): dados em tempo real - (agentes disponíveis/reservados/etc., secao 73), previsão de liberação - via EWMA (secao 74-75), cálculo de pacing (secao 76), CPS distribuído - via Redis (secao 77), reserva atômica de lead (`FOR UPDATE SKIP LOCKED`, - secao 78), lock de campanha com TTL/ownership/renewal (secao 79), - originate via `bgapi` (secao 80), state machine da chamada (secao 82), - controle de abandono (secao 84), regras de retry (secao 86). +- ~~`PredictiveDialerEngine`~~ — implementado na fase CPS Limiter/ + Predictive Engine (ver docs/PREDICTIVE_DIALER.md). - Wizard visual de importação (upload de arquivo de verdade) — fase Frontend. - Relatório de campanha (secao 160) — depende de CDR. diff --git a/docs/PREDICTIVE_DIALER.md b/docs/PREDICTIVE_DIALER.md new file mode 100644 index 0000000..8cbfc11 --- /dev/null +++ b/docs/PREDICTIVE_DIALER.md @@ -0,0 +1,197 @@ +# CPS Limiter / Predictive Dialer Engine + +Agente.md secao 72-86 (motor preditivo) e 77-79 (CPS distribuído, reserva +de leads, lock de campanha). Fecha o discador de verdade — depois desta +fase, uma campanha `RUNNING` origina chamadas sozinha, respeitando +capacidade de agentes, CPS hierárquico e taxa de abandono, sem intervenção +manual. + +Deliberadamente **não** implementado nesta fase (agente.md secao 72: "não é +só `for lead -> originate`"): mod_avmd (secao 87, opcional), callbacks +agendados (secao 88), disposições de agente (secao 89) — ficam pra fase +CDR, que é onde o conceito de "resultado de uma chamada" ganha uma tabela +própria de verdade. + +## Novo serviço: `apps/predictive-dialer` + +Mesmo padrão arquitetural de `apps/freeswitch-events`/`apps/freeswitch- +config`: processo Node standalone em Docker, conexão ESL própria, `tsx` +direto (sem `dist/`), tenant-scoped via `withTenantContext` em toda query. +Um tick a cada 2s: para cada tenant ativo, para cada campanha `RUNNING`/ +`WAITING_SCHEDULE`, tenta processar. + +## Lock de campanha (secao 79) + +`dialer:campaign:{id}` no Redis — `SET NX PX` pra adquirir, renovado a +cada metade do TTL (10s) enquanto o tick daquela campanha está em +andamento, liberado com compare-and-delete (só quem detém o `ownerToken` +consegue renovar/liberar — testado isoladamente: outro dono nunca rouba +nem renova o lock de quem já tem). Garante que, mesmo rodando mais de uma +réplica deste serviço, só um worker processa uma dada campanha por vez. + +## CPS distribuído (secao 77, 62) + +Token bucket por janela fixa de 1s (`INCR` + `PEXPIRE`, script Lua +atômico) — não é sleep(), múltiplos workers batem no mesmo Redis. A +hierarquia (`GLOBAL → TENANT → TRUNK → CAMPAIGN`, node omitido — só existe +um node FreeSWITCH neste deploy) é uma única chamada Lua com todas as +chaves aplicáveis: só incrementa TODAS se TODAS tiverem espaço — testado +isoladamente que um nível esgotado bloqueia mesmo com os outros níveis +tendo espaço de sobra, **sem incrementar parcialmente** os que passariam +(a chave da campanha ficou em 1, não 2, na segunda tentativa bloqueada +pelo tenant). + +## Reserva atômica de leads (secao 78) + +`SELECT ... FOR UPDATE SKIP LOCKED` (raw SQL dentro da mesma transação +`withTenantContext`, RLS + lock de linha coexistindo). `SKIP LOCKED` evita +dois workers reservando o mesmo lead — quem chegar depois pula pro +próximo, nunca espera. `READY`/`NEW` → `RESERVED` no mesmo `UPDATE` dentro +da transação. + +## Dados em tempo real e pacing (secao 73-76) + +`pacing.ts::computeCapacity` conta agentes por estado (via `Tier` → +`Agent.state`, mesma fonte de verdade da fase Agents/Realtime Monitoring) +e estima quantos ficam disponíveis nos próximos 15s — a partir de +`stateUpdatedAt` + `averageTalkTime`/`wrapUpTime`. **Simplificação +deliberada**: a especificação pede 4 buckets de previsão (5/10/15/20s); +aqui é um único horizonte de 15s. Refinar pros 4 buckets fica pra quando +houver dado real suficiente pra validar se faz diferença prática. + +`pacing.ts::decidePacing` aplica o cálculo conceitual da secao 76 +(`expected_agent_capacity = available + predicted`, `calls_to_originate = +round(expected_agent_capacity * pacing_factor / answer_probability) - +calls_in_flight`) e o controle de abandono da secao 84: `abandon_rate` +acima do alvo reduz o pacing; acima de 2x força o mínimo; acima de 3x +suspende originações nesse tick (nunca desliga a campanha sozinha, só pula +o tick). Nunca origina sem capacidade prevista (secao 85: `if +(expectedAgentCapacity <= 0) return 0`). + +## EWMA (secao 75) + +`ewma.ts`, alpha=0.25 — `answer_probability`/`average_answer_delay`/ +`average_talk_time`/`abandon_rate` persistidos em `CampaignStats`, +atualizados a cada tentativa concluída (`call-attempt.ts::completeAttempt`). +Sem Machine Learning (secao 74: "preferir algoritmo estatístico +determinístico e explicável") — é só a fórmula de suavização exponencial, +nada de modelo treinado. + +## Modo simulação (secao 185-186) — como funciona de verdade + +`DIALER_SIMULATION=true` por padrão (já estava no `.env` desde o início da +sessão). Nesse modo, **nenhuma chamada PSTN real acontece**: o desfecho +("o cliente atendeu?") é sorteado inteiramente em software +(`simulation.ts`, 40% ANSWERED / 15% BUSY / 35% NO_ANSWER / 10% FAILED, +com delay e talk time aleatórios) — o FreeSWITCH nem é acionado pra +BUSY/NO_ANSWER/FAILED. + +Quando o sorteio dá **ANSWERED**, aí sim uma chamada real entra no +FreeSWITCH — mas sintética (`null/dummy`, sem PSTN nenhum envolvido), +direto pra `&callcenter(fila@domínio)`. Escolha deliberada: a partir desse +ponto, quem decide o resto (oferecer pro agente, bridgear, abandono por +timeout) é o **mod_callcenter real**, o mesmo já testado e verificado nas +fases Queues/Agents/Realtime Monitoring — maximiza código real exercitado +em vez de simular tudo em memória. Os identificadores da secao 81 +(`b2bcall_tenant_id`/`b2bcall_call_id`/`b2bcall_attempt_id`/ +`b2bcall_campaign_id`/`b2bcall_lead_id`) vão como channel variables nessa +perna sintética — os eventos dela chegam com `tenantId` já resolvido no +WebSocket (fase Realtime Monitoring), sem precisar do fan-out usado por +outros eventos de callcenter. + +**Achado real durante o teste desta fase**: uma perna `null/dummy` não tem +mídia do outro lado — nada faz ela desligar sozinha depois de bridgear com +um agente (diferente de uma ligação de verdade, onde o cliente desliga). +Sem tratar isso, o canal ficava ativo pra sempre. Corrigido: depois de +originar a perna sintética, um `setTimeout` chama `uuid_kill` no +`talk_time` simulado (matando um canal que já terminou sozinho — ex.: +abandonado na fila — não dá erro, sem efeito). + +`event-listener.ts` assina o mesmo canal Redis `b2bcall:events` que a fase +Realtime Monitoring já usa, correlaciona pelo `origination_uuid` +(registry em memória, `queued-attempts-registry.ts`) e fecha o +`CallAttempt` quando `CALL_BRIDGED`/`QUEUE_MEMBER_LEFT`/`CALL_ENDED` +chegam — `AGENT_OFFERED_CALL` só anota qual agente pegou, sem fechar nada +ainda. + +## Real Outbound Safety (secao 186) + +`originateRealPstnLeg` (`originate.ts`) existe, arquiteturalmente completo +(Sofia Gateway real, `sofia/gateway//`), mas só é chamado +quando **as duas** condições da secao 186 são verdadeiras +(`DIALER_SIMULATION=false` E `ALLOW_REAL_OUTBOUND_CALLS=true`) — nunca +ativado automaticamente, checado uma vez no boot do worker +(`main.ts::readOutboundSafetyFlags`, log de warning claro em cada estado). +**Nunca exercitado nesta sessão** — não existe trunk/operadora real +disponível neste laboratório. O caminho de `answer_delay` real (via evento +`CALL_ANSWERED`, distinto do delay pré-sorteado do modo simulação) também +está implementado mas não testado pelo mesmo motivo. + +## Achado real: corrida entre `queue:sync` e `tier:sync` + +Descoberto testando esta fase: criar uma fila e atribuir um tier logo em +seguida (fluxo normal de setup de uma campanha) pode fazer `tier add` +rodar **antes** do `queue reload` da fila terminar no FreeSWITCH — erro +real (`-ERR Queue not found!`), diferente do já conhecido "already exist" +(esse sim inofensivo). Como `queues:sync` e `tiers:sync` são dois canais +Redis independentes sem ordem garantida entre si, o login do agente (que +já resincroniza tiers, ver docs/AGENTS.md) não bastava — os dois disparos +corriam antes do reload terminar. Corrigido com retry curto (até 3 +tentativas, backoff de 500ms/1s/1.5s) especificamente pra esse erro, em +`apps/freeswitch-config/src/agent-sync.ts::addTierWithRetry`. + +## `GET /campaigns/:id/stats` (secao 227.7 "visualizar pacing") + +Expõe `CampaignStats` (atualizado a cada tick) + contagem de agentes por +estado + chamadas em andamento, calculado na hora. Suficiente pro +acceptance criteria "visualizar pacing" sem esperar a fase Frontend. + +## Verificado ponta a ponta + +``` +Campanha RUNNING, 1 agente disponível, 3 leads: + tick -> originando 3 tentativas (pacing_factor=1.05, answer_probability=0.4) + Lead 1: FAILED (simulado) -> retry em 30min + Lead 2: NO_ANSWER (simulado) -> retry em 60min + Lead 3: ANSWERED (simulado) -> entra na fila real -> AGENT_OFFERED_CALL + -> CALL_BRIDGED (agente conectado, evento real) -> hangup agendado + -> CALL_ENDED -> CallAttempt COMPLETED, EWMA atualizada + +Corrida queue/tier: "tier add" falhando com "Queue not found" -> retry + automático -> "tier sincronizado" -> tier list confirma o agente na fila + +CPS limiter isolado: maxPerSecond=2, 4 tentativas -> true,true,false,false +CPS hierarquia: nível tenant esgotado bloqueia mesmo com campanha livre, + SEM incremento parcial (contador da campanha ficou em 1, não 2) +Lock de campanha isolado: dono B nunca rouba nem renova lock do dono A; + dono C adquire normalmente depois do release do dono A + +stop numa campanha com chamada ativa -> não derruba a chamada em curso + (secao 66), só para novas originações — confirmado observando o canal + continuar ativo até seu próprio hangup agendado + +queue list mostra calls_answered=3 ao final dos testes, confirmando 3 +conexões reais de agente via o pipeline completo +``` + +typecheck do workspace inteiro limpo. + +## O que falta + +- Previsão de 4 buckets (5/10/15/20s, secao 74) — simplificado pra um + único horizonte de 15s. +- `mod_avmd` (secao 87), callbacks agendados (secao 88), disposições de + agente (secao 89) — fora do escopo desta fase. +- Caminho PSTN real (secao 80-83, 186) — implementado, nunca exercitado + (sem trunk real disponível). +- CPS a nível de trunk (`Trunk.maxCps`) — campo existe no modelo desde a + fase Trunks, mas só é aplicado quando o caminho PSTN real rodar (a perna + simulada não passa por um trunk de verdade, não faz sentido contar CPS + de trunk pra ela). +- Memória do worker ficou em ~266MB depois de alguns ciclos de teste + (contra ~30-55MB de fs-events/fs-config) — dentro do orçamento da VM, + mas vale reavaliar se crescer mais rodando por mais tempo/mais + campanhas simultâneas. +- Relatório de campanha (secao 160), TME/TMA/Service Level/Abandon Rate + agregados de verdade — dependem de CDR (próxima fase da ordem do + agente.md secao 232). diff --git a/packages/database/prisma/migrations/20260828155954_call_attempts_campaign_stats/migration.sql b/packages/database/prisma/migrations/20260828155954_call_attempts_campaign_stats/migration.sql new file mode 100644 index 0000000..271ae0e --- /dev/null +++ b/packages/database/prisma/migrations/20260828155954_call_attempts_campaign_stats/migration.sql @@ -0,0 +1,60 @@ +-- CreateEnum +CREATE TYPE "call_attempt_status" AS ENUM ('CREATED', 'RESERVED', 'ORIGINATING', 'ORIGINATED', 'RINGING', 'ANSWERED', 'QUEUEING', 'AGENT_CONNECTED', 'COMPLETED', 'BUSY', 'NO_ANSWER', 'FAILED', 'ABANDONED'); + +-- CreateTable +CREATE TABLE "call_attempts" ( + "id" UUID NOT NULL, + "tenant_id" UUID NOT NULL, + "campaign_id" UUID NOT NULL, + "lead_id" UUID NOT NULL, + "agent_id" UUID, + "origination_uuid" UUID, + "status" "call_attempt_status" NOT NULL DEFAULT 'CREATED', + "simulated" BOOLEAN NOT NULL DEFAULT false, + "hangup_cause" TEXT, + "talk_time_seconds" INTEGER, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "ringing_at" TIMESTAMP(3), + "answered_at" TIMESTAMP(3), + "bridged_at" TIMESTAMP(3), + "ended_at" TIMESTAMP(3), + + CONSTRAINT "call_attempts_pkey" PRIMARY KEY ("id") +); + +-- CreateTable +CREATE TABLE "campaign_stats" ( + "campaign_id" UUID NOT NULL, + "answer_probability" DOUBLE PRECISION NOT NULL DEFAULT 0.4, + "average_answer_delay" DOUBLE PRECISION NOT NULL DEFAULT 5, + "average_talk_time" DOUBLE PRECISION NOT NULL DEFAULT 180, + "abandon_rate" DOUBLE PRECISION NOT NULL DEFAULT 0, + "pacing_factor" DOUBLE PRECISION NOT NULL DEFAULT 1.0, + "updated_at" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "campaign_stats_pkey" PRIMARY KEY ("campaign_id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "call_attempts_origination_uuid_key" ON "call_attempts"("origination_uuid"); + +-- CreateIndex +CREATE INDEX "call_attempts_tenant_id_idx" ON "call_attempts"("tenant_id"); + +-- CreateIndex +CREATE INDEX "call_attempts_campaign_id_status_idx" ON "call_attempts"("campaign_id", "status"); + +-- AddForeignKey +ALTER TABLE "call_attempts" ADD CONSTRAINT "call_attempts_tenant_id_fkey" FOREIGN KEY ("tenant_id") REFERENCES "tenants"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "call_attempts" ADD CONSTRAINT "call_attempts_campaign_id_fkey" FOREIGN KEY ("campaign_id") REFERENCES "campaigns"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "call_attempts" ADD CONSTRAINT "call_attempts_lead_id_fkey" FOREIGN KEY ("lead_id") REFERENCES "leads"("id") ON DELETE RESTRICT ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "call_attempts" ADD CONSTRAINT "call_attempts_agent_id_fkey" FOREIGN KEY ("agent_id") REFERENCES "agents"("id") ON DELETE SET NULL ON UPDATE CASCADE; + +-- AddForeignKey +ALTER TABLE "campaign_stats" ADD CONSTRAINT "campaign_stats_campaign_id_fkey" FOREIGN KEY ("campaign_id") REFERENCES "campaigns"("id") ON DELETE RESTRICT ON UPDATE CASCADE; diff --git a/packages/database/prisma/migrations/20260828160116_add_tenant_to_campaign_stats/migration.sql b/packages/database/prisma/migrations/20260828160116_add_tenant_to_campaign_stats/migration.sql new file mode 100644 index 0000000..46fe6ec --- /dev/null +++ b/packages/database/prisma/migrations/20260828160116_add_tenant_to_campaign_stats/migration.sql @@ -0,0 +1,14 @@ +/* + Warnings: + + - Added the required column `tenant_id` to the `campaign_stats` table without a default value. This is not possible if the table is not empty. + +*/ +-- AlterTable +ALTER TABLE "campaign_stats" ADD COLUMN "tenant_id" UUID NOT NULL; + +-- CreateIndex +CREATE INDEX "campaign_stats_tenant_id_idx" ON "campaign_stats"("tenant_id"); + +-- AddForeignKey +ALTER TABLE "campaign_stats" ADD CONSTRAINT "campaign_stats_tenant_id_fkey" FOREIGN KEY ("tenant_id") REFERENCES "tenants"("id") ON DELETE RESTRICT ON UPDATE CASCADE; diff --git a/packages/database/prisma/migrations/20260828160500_call_attempts_campaign_stats_rls/migration.sql b/packages/database/prisma/migrations/20260828160500_call_attempts_campaign_stats_rls/migration.sql new file mode 100644 index 0000000..a4403ac --- /dev/null +++ b/packages/database/prisma/migrations/20260828160500_call_attempts_campaign_stats_rls/migration.sql @@ -0,0 +1,10 @@ +-- Tabelas de negocio tenant-scoped: RLS obrigatorio em todas (ver docs/TENANT_ISOLATION.md). +ALTER TABLE "call_attempts" ENABLE ROW LEVEL SECURITY; +ALTER TABLE "call_attempts" FORCE ROW LEVEL SECURITY; +CREATE POLICY "tenant_isolation" ON "call_attempts" + USING (tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid); + +ALTER TABLE "campaign_stats" ENABLE ROW LEVEL SECURITY; +ALTER TABLE "campaign_stats" FORCE ROW LEVEL SECURITY; +CREATE POLICY "tenant_isolation" ON "campaign_stats" + USING (tenant_id = NULLIF(current_setting('app.current_tenant_id', true), '')::uuid); diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index ab0364c..d7e391c 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -50,6 +50,8 @@ model Tenant { campaigns Campaign[] leads Lead[] suppressionEntries SuppressionEntry[] + callAttempts CallAttempt[] + campaignStats CampaignStats[] @@map("tenants") } @@ -532,13 +534,14 @@ model Agent { updatedAt DateTime @updatedAt @map("updated_at") deletedAt DateTime? @map("deleted_at") - tenant Tenant @relation(fields: [tenantId], references: [id]) - user User @relation(fields: [userId], references: [id]) - extension Extension? @relation(fields: [extensionId], references: [id]) - tiers Tier[] - sessions AgentSession[] - stateEvents AgentStateEvent[] - pauseEvents AgentPauseEvent[] + tenant Tenant @relation(fields: [tenantId], references: [id]) + user User @relation(fields: [userId], references: [id]) + extension Extension? @relation(fields: [extensionId], references: [id]) + tiers Tier[] + sessions AgentSession[] + stateEvents AgentStateEvent[] + pauseEvents AgentPauseEvent[] + callAttempts CallAttempt[] @@unique([tenantId, userId]) @@index([tenantId]) @@ -711,10 +714,12 @@ model Campaign { updatedAt DateTime @updatedAt @map("updated_at") deletedAt DateTime? @map("deleted_at") - tenant Tenant @relation(fields: [tenantId], references: [id]) - queue Queue @relation(fields: [queueId], references: [id]) - trunk Trunk @relation(fields: [trunkId], references: [id]) - leads Lead[] + tenant Tenant @relation(fields: [tenantId], references: [id]) + queue Queue @relation(fields: [queueId], references: [id]) + trunk Trunk @relation(fields: [trunkId], references: [id]) + leads Lead[] + callAttempts CallAttempt[] + campaignStats CampaignStats[] @@unique([tenantId, name]) @@index([tenantId]) @@ -776,8 +781,9 @@ model Lead { createdAt DateTime @default(now()) @map("created_at") updatedAt DateTime @updatedAt @map("updated_at") - tenant Tenant @relation(fields: [tenantId], references: [id]) - campaign Campaign @relation(fields: [campaignId], references: [id]) + tenant Tenant @relation(fields: [tenantId], references: [id]) + campaign Campaign @relation(fields: [campaignId], references: [id]) + callAttempts CallAttempt[] @@unique([campaignId, phoneNormalized]) @@index([tenantId]) @@ -803,3 +809,93 @@ model SuppressionEntry { @@index([tenantId]) @@map("suppression_entries") } + +enum CallAttemptStatus { + CREATED + RESERVED + ORIGINATING + ORIGINATED + RINGING + ANSWERED + QUEUEING + AGENT_CONNECTED + COMPLETED + BUSY + NO_ANSWER + FAILED + ABANDONED + + @@map("call_attempt_status") +} + +// Uma tentativa de discagem de um lead (agente.md secao 81-82). Os +// identificadores daqui (id, campaignId, leadId, originationUuid) viram +// channel variables b2bcall_* no originate (secao 81) — é assim que +// b2bcall-fs-events consegue popular NormalizedEvent.tenantId/b2bcallCallId +// pra chamadas do discador, diferente das chamadas internas (extension a +// extension) que não carregam esses vars ainda. +model CallAttempt { + id String @id @default(uuid()) @db.Uuid + tenantId String @map("tenant_id") @db.Uuid + campaignId String @map("campaign_id") @db.Uuid + leadId String @map("lead_id") @db.Uuid + agentId String? @map("agent_id") @db.Uuid + + // UUID do channel no FreeSWITCH (real ou simulado) — null até o originate + // de fato rodar (status ainda CREATED/RESERVED). + originationUuid String? @unique @map("origination_uuid") @db.Uuid + + status CallAttemptStatus @default(CREATED) + + // agente.md secao 185: em modo simulação, nenhuma chamada PSTN real + // acontece — answer/busy/no_answer/failed/ringing/delay/talk_time são + // sorteados em software, nunca originados de verdade pro trunk. + simulated Boolean @default(false) + + hangupCause String? @map("hangup_cause") + talkTimeSeconds Int? @map("talk_time_seconds") + + createdAt DateTime @default(now()) @map("created_at") + ringingAt DateTime? @map("ringing_at") + answeredAt DateTime? @map("answered_at") + bridgedAt DateTime? @map("bridged_at") + endedAt DateTime? @map("ended_at") + + tenant Tenant @relation(fields: [tenantId], references: [id]) + campaign Campaign @relation(fields: [campaignId], references: [id]) + lead Lead @relation(fields: [leadId], references: [id]) + agent Agent? @relation(fields: [agentId], references: [id]) + + @@index([tenantId]) + @@index([campaignId, status]) + @@map("call_attempts") +} + +// Estatísticas EWMA por campanha (agente.md secao 73-76), consultadas e +// atualizadas a cada tick do PredictiveDialerEngine — persistidas pra +// sobreviver a reinícios do worker (não é cache descartável). Valores +// iniciais são estimativas conservadoras (secao 74: "não precisa Machine +// Learning, preferir algoritmo estatístico determinístico e explicável"), +// convergem conforme tentativas reais completam. +model CampaignStats { + campaignId String @id @map("campaign_id") @db.Uuid + tenantId String @map("tenant_id") @db.Uuid + + answerProbability Float @default(0.4) @map("answer_probability") + averageAnswerDelay Float @default(5) @map("average_answer_delay") + averageTalkTime Float @default(180) @map("average_talk_time") + abandonRate Float @default(0) @map("abandon_rate") + + // Fator de pacing aplicado no cálculo de quantas chamadas originar por + // tick (secao 76) — começa em Campaign.pacingInitial, ajustado pelo + // controle de abandono (secao 84), sempre dentro de [pacingMin, pacingMax]. + pacingFactor Float @default(1.0) @map("pacing_factor") + + updatedAt DateTime @updatedAt @map("updated_at") + + tenant Tenant @relation(fields: [tenantId], references: [id]) + campaign Campaign @relation(fields: [campaignId], references: [id]) + + @@index([tenantId]) + @@map("campaign_stats") +} diff --git a/packages/telephony/src/normalize-event.ts b/packages/telephony/src/normalize-event.ts index aaffe6e..8c0f47e 100644 --- a/packages/telephony/src/normalize-event.ts +++ b/packages/telephony/src/normalize-event.ts @@ -11,6 +11,7 @@ function baseFields(headers: RawHeaders, extra: Record = {}) { callUuid: headers["Unique-ID"], tenantId: channelVar(headers, "b2bcall_tenant_id"), b2bcallCallId: channelVar(headers, "b2bcall_call_id"), + b2bcallAttemptId: channelVar(headers, "b2bcall_attempt_id"), b2bcallCampaignId: channelVar(headers, "b2bcall_campaign_id"), b2bcallLeadId: channelVar(headers, "b2bcall_lead_id"), data: { ...extra }, diff --git a/packages/telephony/src/types.ts b/packages/telephony/src/types.ts index 99cbcef..33fced3 100644 --- a/packages/telephony/src/types.ts +++ b/packages/telephony/src/types.ts @@ -70,10 +70,12 @@ export interface NormalizedEvent { occurredAt: string; /** UUID do channel/call quando aplicável. */ callUuid?: string; - /** Channel variables b2bcall_* quando presentes (secao 81) — ainda não - * populadas nesta fase (só existirão a partir do Predictive Engine). */ + /** Channel variables b2bcall_* (secao 81) — populadas pra chamadas + * originadas pelo PredictiveDialerEngine (fase Predictive Engine); + * chamadas internas (extensão a extensão) ainda não carregam esses vars. */ tenantId?: string; b2bcallCallId?: string; + b2bcallAttemptId?: string; b2bcallCampaignId?: string; b2bcallLeadId?: string; data: Record; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 585f8cd..3b0d2d5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -147,6 +147,34 @@ importers: specifier: ^5.7.0 version: 5.9.3 + apps/predictive-dialer: + dependencies: + '@b2bcall/database': + specifier: workspace:* + version: link:../../packages/database + '@b2bcall/shared': + specifier: workspace:* + version: link:../../packages/shared + '@b2bcall/telephony': + specifier: workspace:* + version: link:../../packages/telephony + esl: + specifier: 11.2.1 + version: 11.2.1 + ioredis: + specifier: ^6.0.0 + version: 6.0.0 + devDependencies: + '@types/node': + specifier: ^22.0.0 + version: 22.20.1 + tsx: + specifier: ^4.23.12 + version: 4.23.12 + typescript: + specifier: ^5.7.0 + version: 5.9.3 + packages/auth: dependencies: '@b2bcall/database':