feat(dialer): CPS Limiter + Predictive Dialer Engine
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01X1HxY46WGU4G1zmVDNKcWw
This commit is contained in:
244
apps/predictive-dialer/src/tick.ts
Normal file
244
apps/predictive-dialer/src/tick.ts
Normal file
@@ -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<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
export async function tickCampaign(deps: TickDeps, tenant: Tenant, campaign: Campaign): Promise<void> {
|
||||
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<number | null> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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 }));
|
||||
}
|
||||
Reference in New Issue
Block a user